--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 8fe8ef447bb4716f00abd033149cf2ce96839fb4
Parents : 26d5a3d
Author : Mark Qvist <bc7291552be7a58f361522990465165c>
Signature : T66BB85Valid, signed by author
Date : 2026-07-14T23:01:19+02:00
Added interactive console
Changes
41 files changed, 3238 insertions(+), 111 deletions(-)
Diff
diff --git a/sbapp/__init__.py b/sbapp/__init__.py
index f49e360a..49612e76 100644
--- a/sbapp/__init__.py
+++ b/sbapp/__init__.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
import glob
diff --git a/sbapp/gv.py b/sbapp/gv.py
index 6d07c8c0..ebf2eedc 100644
--- a/sbapp/gv.py
+++ b/sbapp/gv.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import re
import os
def gv() -> str:
diff --git a/sbapp/main.py b/sbapp/main.py
index 309d1dda..839dbfef 100644
--- a/sbapp/main.py
+++ b/sbapp/main.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
__debug_build__ = False
__disable_shaders__ = False
__version__ = "2.0.0"
@@ -526,6 +528,7 @@ class SidebandApp(MDApp):
self.app_dir = plyer.storagepath.get_application_dir()
self.shaders_disabled = __disable_shaders__
self.keyboard_enabled = False
+ self.console_attached = False
self.no_transition = NoTransition()
self.slide_transition = SlideTransition()
@@ -545,13 +548,9 @@ class SidebandApp(MDApp):
if args.interactive:
def cj():
while not self.sideband.getstate("core.started") == True: time.sleep(0.1)
- import importlib
- if importlib.util.find_spec('prompt_toolkit') != None:
- from .sideband import console
- time.sleep(0.7); console.attach(self.sideband, full_screen=True)
- else:
- print("Could not start Sideband console, since the \"prompt-toolkit\" module is not available")
- print("You can install it with \"pip install prompt-toolkit\"")
+ from .sideband import console
+ self.console_attached = True
+ time.sleep(1.5); console.attach(self.sideband, history_path=f"{self.sideband.app_dir}/console.history")
threading.Thread(target=cj, daemon=True).start()
@@ -1927,7 +1926,16 @@ class SidebandApp(MDApp):
self.root.ids.screen_manager.current = "loader_screen"
self.root.ids.nav_drawer.set_state("closed")
+ def schedule_quit(self):
+ Clock.schedule_once(self.quit_action, 0.1)
+
def quit_action(self, sender):
+ if self.console_attached:
+ from .sideband import console
+ console.detach()
+ self.console_attached = False
+ time.sleep(0.2)
+
self.closing_app = True
self.root.ids.nav_drawer.set_state("closed")
self.sideband.should_persist_data()
@@ -5217,14 +5225,9 @@ def run():
if args.interactive:
while not sideband.getstate("core.started") == True: time.sleep(0.1)
- import importlib
- if importlib.util.find_spec('prompt_toolkit') != None:
- from .sideband import console
- console.attach(sideband)
- else:
- print("Could not start Sideband console, since the \"prompt-toolkit\" module is not available")
- print("You can install it with \"pip install prompt-toolkit\"")
-
+ from .sideband import console
+ console.attach(sideband, history_path=f"{sideband.app_dir}/console.history")
+
else:
while True: time.sleep(5)
else:
diff --git a/sbapp/services/sidebandservice.py b/sbapp/services/sidebandservice.py
index e5a4b438..2ca5e9ef 100644
--- a/sbapp/services/sidebandservice.py
+++ b/sbapp/services/sidebandservice.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
__debug_build__ = False
import sys
diff --git a/sbapp/sideband/__init__.py b/sbapp/sideband/__init__.py
index f49e360a..49612e76 100644
--- a/sbapp/sideband/__init__.py
+++ b/sbapp/sideband/__init__.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
import glob
diff --git a/sbapp/sideband/audioproc.py b/sbapp/sideband/audioproc.py
index 215fcd87..fc4530ac 100644
--- a/sbapp/sideband/audioproc.py
+++ b/sbapp/sideband/audioproc.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
import io
import math
diff --git a/sbapp/sideband/cli/__init__.py b/sbapp/sideband/cli/__init__.py
new file mode 100644
index 00000000..2aa6adce
--- /dev/null
+++ b/sbapp/sideband/cli/__init__.py
@@ -0,0 +1,3 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+from .commands import Command, CommandRegistry, registry
\ No newline at end of file
diff --git a/sbapp/sideband/cli/cmd_config.py b/sbapp/sideband/cli/cmd_config.py
new file mode 100644
index 00000000..2df7315e
--- /dev/null
+++ b/sbapp/sideband/cli/cmd_config.py
@@ -0,0 +1,185 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import RNS
+from .commands import Command, registry
+from . import output as out
+from . import config_tree as ct
+
+def _cmd_config(args, sideband, ctx):
+ categories = ct.get_categories()
+ lines = []
+ lines.append(out.section("Configuration Categories"))
+ lines.append("Use 'config list <category>' to list aliases in a category.\n")
+ for cat in categories:
+ alias_count = len(ct.get_aliases(cat))
+ lines.append(f" {cat:<16s} ({alias_count} aliases)")
+ lines.append(f" {'total':<16s} ({len(ct.get_all_keys())} keys)")
+ return "\n".join(lines)
+
+def _cmd_config_list(args, sideband, ctx):
+ if not args:
+ lines = []
+ lines.append(out.section("All Configuration Entries"))
+ for cat in ct.get_categories():
+ lines.append(f" [{cat}]")
+ for alias in ct.get_aliases(cat):
+ flat_key = ct.resolve_alias(cat, alias)
+ value = sideband.config.get(flat_key)
+ info = ct.get_key_info(flat_key)
+ vtype = info["type"] if info else "?"
+ formatted = ct.format_value(flat_key, value)
+ lines.append(f" {alias:<30s} {formatted}")
+ lines.append("")
+ return "\n".join(lines)
+
+ category = args[0]
+ aliases = ct.get_aliases(category)
+ if not aliases:
+ # Maybe typed a flat key name by mistake
+ if category in ct.get_all_keys():
+ return f"'{category}' is a flat config key.\nUse 'config get <category> <alias>' or 'raw sideband.config[\"{category}\"]'"
+ return f"Unknown category: {category}\nAvailable categories: {', '.join(ct.get_categories())}"
+
+ lines = []
+ lines.append(out.section(f"Configuration: {category}"))
+ for alias in aliases:
+ flat_key = ct.resolve_alias(category, alias)
+ value = sideband.config.get(flat_key)
+ info = ct.get_key_info(flat_key)
+ vtype = info["type"] if info else "?"
+ desc = info["description"] if info else ""
+ formatted = ct.format_value(flat_key, value)
+ lines.append(f" {alias:<30s} {vtype:<6s} {formatted}")
+ return "\n".join(lines)
+
+def _cmd_config_get(args, sideband, ctx):
+ if len(args) < 2: return "Usage: config get <category> <alias>"
+
+ category = args[0]
+ alias = args[1]
+ flat_key = ct.resolve_alias(category, alias)
+
+ if flat_key is None: return f"Unknown alias: '{alias}' in category '{category}'\nUse 'config list {category}' to see available aliases."
+ if flat_key not in sideband.config: return f"Config key '{flat_key}' not present in active configuration."
+
+ value = sideband.config[flat_key]
+ info = ct.get_key_info(flat_key)
+ type_str = f" ({info['type']})" if info else ""
+
+ lines = []
+ lines.append(f"{category} {alias} = {ct.format_value(flat_key, value)}{type_str}")
+ return "\n".join(lines)
+
+def _cmd_config_set(args, sideband, ctx):
+ if len(args) < 3: return "Usage: config set <category> <alias> <value>"
+
+ category = args[0]
+ alias = args[1]
+ raw_value = " ".join(args[2:]) if len(args) > 2 else ""
+
+ flat_key = ct.resolve_alias(category, alias)
+ if flat_key is None:
+ return f"Unknown alias: '{alias}' in category '{category}'\nUse 'config list {category}' to see available aliases."
+
+ if flat_key not in sideband.config:
+ info = ct.get_key_info(flat_key)
+ if info is None: return f"Unknown config key: {flat_key}"
+
+ old_value = sideband.config.get(flat_key)
+
+ try: new_value = ct.convert_value(flat_key, raw_value)
+ except ValueError as e: return f"Error: {e}"
+
+ sideband.config[flat_key] = new_value
+
+ lines = []
+ lines.append(f"{category} {alias}: {ct.format_value(flat_key, old_value)} -> {ct.format_value(flat_key, new_value)}")
+ _cmd_config_save(args, sideband, ctx)
+ return "\n".join(lines)
+
+def _cmd_config_save(args, sideband, ctx):
+ try:
+ sideband.save_configuration()
+ return "Configuration saved"
+ except Exception as e: return f"Error saving configuration: {e}"
+
+def _cmd_config_reload(args, sideband, ctx):
+ try:
+ sideband.reload_configuration()
+ return "Configuration reloaded"
+ except Exception as e: return f"Error reloading configuration: {e}"
+
+def register(sideband=None):
+ registry.register(Command(
+ path=("config",),
+ help_text="Show configuration categories",
+ help_detail="""config
+ Lists all configuration categories. Use 'config list <category>'
+ to see aliases within a category, or 'config get <category> <alias>'
+ to read a value.""",
+ max_args=0,
+ handler=_cmd_config,
+ ))
+
+ registry.register(Command(
+ path=("config", "list"),
+ help_text="List configuration aliases",
+ help_detail="""config list [category]
+ Lists all configuration aliases, or aliases within a specific category.
+ Each alias is shown with its current value, type, and description.""",
+ max_args=1,
+ handler=_cmd_config_list,
+ ))
+
+ registry.register(Command(
+ path=("config", "get"),
+ help_text="Get a configuration value",
+ help_detail="""config get <category> <alias>
+ Displays the current value, type, and description of the
+ specified configuration alias.
+
+ <category> Configuration category (e.g. sensors, telemetry)
+ <alias> Alias within the category (e.g. humidity, mqtt_host)""",
+ min_args=2,
+ max_args=2,
+ handler=_cmd_config_get,
+ ))
+
+ registry.register(Command(
+ path=("config", "set"),
+ help_text="Set a configuration value",
+ help_detail="""config set <category> <alias> <value>
+ Sets a configuration alias to the specified value. The value is
+ automatically converted to the correct type based on the alias's
+ metadata (bool, int, float, str, list).
+
+ Boolean values accept: true/false, yes/no, on/off, 1/0
+ None values accept: none, null, (empty string for str keys)
+
+ <category> Configuration category
+ <alias> Alias within the category
+ <value> New value""",
+ min_args=2,
+ handler=_cmd_config_set,
+ ))
+
+ registry.register(Command(
+ path=("config", "save"),
+ help_text="Persist configuration to disk",
+ help_detail="""config save
+ Writes the current in-memory configuration to the config file
+ on disk.""",
+ max_args=0,
+ handler=_cmd_config_save,
+ ))
+
+ registry.register(Command(
+ path=("config", "reload"),
+ help_text="Reload configuration from disk",
+ help_detail="""config reload
+ Reloads the configuration from the on-disk config file, discarding
+ any unsaved in-memory changes. Some settings may require a full
+ restart to take full effect (e.g. interface configuration).""",
+ max_args=0,
+ handler=_cmd_config_reload,
+ ))
diff --git a/sbapp/sideband/cli/cmd_conv.py b/sbapp/sideband/cli/cmd_conv.py
new file mode 100644
index 00000000..484ad419
--- /dev/null
+++ b/sbapp/sideband/cli/cmd_conv.py
@@ -0,0 +1,369 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import re
+import RNS
+from .commands import Command, registry
+from . import output as out
+
+def _resolve_dest(sideband, dest_str: str):
+ dest_str = dest_str.strip()
+ try:
+ dest = bytes.fromhex(dest_str)
+ if len(dest) == RNS.Reticulum.TRUNCATED_HASHLENGTH // 8:
+ name = sideband.peer_display_name(dest)
+ return dest, name
+ except ValueError: pass
+
+ try:
+ for conv in sideband.list_conversations():
+ name = out.sanitize_name(sideband.peer_display_name(conv["dest"]))
+ if name and name.lower() == out.sanitize_name(dest_str).lower(): return conv["dest"], name
+ except Exception: pass
+
+ return None, None
+
+def _cmd_conv(args, sideband, ctx):
+ conversations = sideband.list_conversations(objects=True)
+ if not conversations: return "No conversations"
+
+ lines = []
+ lines.append(out.section("Conversations"))
+ lines.append(f" {'Name':<24s} {'Hash':<34s} {'Trust':>6s} {'Unread':>7s} {'Last Activity':>14s}")
+ lines.append(f" {'-'*24} {'-'*34} {'-'*6} {'-'*7} {'-'*14}")
+
+ for conv in conversations:
+ dest = conv["dest"]
+ name = out.sanitize_name(sideband.peer_display_name(dest))
+ name = name[:23]
+ hex_str = RNS.prettyhexrep(dest)
+ trust = "yes" if conv.get("trust", 0) else "no"
+ unread = str(conv.get("unread", 0))
+ last = out.time_ago(conv.get("last_activity"))
+ lines.append(f" {name:<24s} {hex_str:<24s} {trust:>6s} {unread:>7s} {last:>14s}")
+
+ return "\n".join(lines)
+
+def _cmd_conv_send(args, sideband, ctx):
+ dest_str = args[0]
+ message = " ".join(args[1:])
+
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+
+ propagation = sideband.config.get("propagation_by_default", False)
+ result = sideband.send_message(message, dest, propagation)
+ if result: return f"Message sent to {name or RNS.prettyhexrep(dest)}"
+ else: return f"Failed to send message to {name or RNS.prettyhexrep(dest)}"
+
+def _cmd_conv_command(args, sideband, ctx):
+ dest_str = args[0]
+ command = " ".join(args[1:])
+
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+
+ propagation = sideband.config.get("propagation_by_default", False)
+ result = sideband.send_command(command, dest, propagation)
+ if result: return f"Command sent to {name or RNS.prettyhexrep(dest)}"
+ else: return f"Failed to send command to {name or RNS.prettyhexrep(dest)}"
+
+
+def _cmd_conv_announce(args, sideband, ctx):
+ sideband.lxmf_announce()
+ return "LXMF announce sent."
+
+def _cmd_conv_sync(args, sideband, ctx):
+ limit = None
+ if args:
+ try: limit = int(args[0])
+ except ValueError: return f"Invalid sync limit: {args[0]}"
+
+ result = sideband.request_lxmf_sync(limit)
+ if result: return f"LXMF sync requested{f' (limit {limit})' if limit else ''}."
+ else: return "Sync already in progress or no propagation node configured"
+
+def _cmd_conv_sync_cancel(args, sideband, ctx):
+ sideband.cancel_lxmf_sync()
+ return "LXMF sync cancelled"
+
+def _cmd_conv_sync_status(args, sideband, ctx):
+ status = sideband.get_sync_status()
+ progress = sideband.get_sync_progress()
+ return f"Sync status: {status}\nProgress: {int(progress * 100)}%"
+
+def _cmd_conv_msgs(args, sideband, ctx):
+ dest_str = args[0]
+ count = 12
+ if len(args) > 1:
+ try: count = int(args[1])
+ except ValueError: return f"Invalid message count: {args[1]}"
+
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+
+ messages = sideband.list_messages(dest, limit=count)
+ if not messages: return f"No messages in conversation with {name or RNS.prettyhexrep(dest)}."
+
+ lines = []
+ lines.append(out.section(f"Messages: {name or RNS.prettyhexrep(dest)}"))
+
+ for msg in messages:
+ direction = "out" if msg["source"] == sideband.lxmf_destination.hash else "in"
+ state_str = _format_msg_state(msg["state"], msg["method"]) if direction == "out" else "received"
+ ts = out.timestamp(msg["received"])
+ content = msg["content"].decode("utf-8").strip()
+ if len(content) < 1: content = "No text content in message"
+ size_str = f" ({RNS.prettysize(msg['lxm'].packed_size)})" if msg["lxm"] else ""
+ lines.append(f"[{direction}]{' ' if direction == 'in' else ''} {ts} {state_str}{size_str}\n{content}\n")
+
+ return "\n".join(lines)
+
+def _format_msg_state(state, method):
+ import LXMF
+ if state == LXMF.LXMessage.DELIVERED: return "delivered"
+ elif state == LXMF.LXMessage.FAILED: return "failed"
+ elif state == LXMF.LXMessage.CANCELLED: return "cancelled"
+ elif state == LXMF.LXMessage.REJECTED: return "rejected"
+ elif method == LXMF.LXMessage.PROPAGATED and state == LXMF.LXMessage.SENT: return "propagated"
+ elif method == LXMF.LXMessage.PAPER: return "paper"
+ elif state == LXMF.LXMessage.OUTBOUND or state == LXMF.LXMessage.SENDING: return "sending"
+ elif state == LXMF.LXMessage.SENT: return "sent"
+ else: return f"unknown ({state})"
+
+def _cmd_conv_create(args, sideband, ctx):
+ dest_str = args[0]
+ name = args[1] if len(args) > 1 else ""
+
+ try:
+ dest = bytes.fromhex(dest_str)
+ if len(dest) != RNS.Reticulum.TRUNCATED_HASHLENGTH // 8:
+ return f"Invalid destination hash: must be {RNS.Reticulum.TRUNCATED_HASHLENGTH // 8 * 2} hex characters"
+ except ValueError: return f"Invalid destination hash: {dest_str!r}"
+
+ if sideband.has_conversation(dest): return f"Conversation with {RNS.prettyhexrep(dest)} already exists"
+
+ result = sideband.new_conversation(dest_str, name, trusted=False)
+ if result: return f"Conversation created with {RNS.prettyhexrep(dest)}{f' ({name})' if name else ''}"
+ else: return f"Failed to create conversation with {RNS.prettyhexrep(dest)}"
+
+
+def _cmd_conv_delete(args, sideband, ctx):
+ dest_str = args[0]
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+ sideband.delete_conversation(dest)
+ return f"Conversation with {name or RNS.prettyhexrep(dest)} deleted"
+
+def _cmd_conv_clear(args, sideband, ctx):
+ dest_str = args[0]
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+
+ sideband.clear_conversation(dest)
+ return f"Messages cleared for conversation with {name or RNS.prettyhexrep(dest)}"
+
+def _cmd_conv_trust(args, sideband, ctx):
+ dest_str = args[0]
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+ sideband.trusted_conversation(dest)
+ return f"Conversation with {name or RNS.prettyhexrep(dest)} marked as trusted"
+
+
+def _cmd_conv_untrust(args, sideband, ctx):
+ dest_str = args[0]
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+ sideband.untrusted_conversation(dest)
+ return f"Conversation with {name or RNS.prettyhexrep(dest)} marked as untrusted"
+
+def _cmd_conv_name(args, sideband, ctx):
+ dest_str = args[0]
+ name = " ".join(args[1:])
+ dest, _ = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+ sideband.named_conversation(name, dest)
+ return f"Display name set to '{name}' for {RNS.prettyhexrep(dest)}"
+
+def _cmd_conv_paper(args, sideband, ctx):
+ dest_str = args[0]
+ message = " ".join(args[1:])
+ dest, name = _resolve_dest(sideband, dest_str)
+ if dest is None: return f"Unknown destination: {dest_str}"
+ result = sideband.paper_message(message, dest)
+ if result: return f"Paper message created for {name or RNS.prettyhexrep(dest)}"
+ else: return f"Failed to create paper message for {name or RNS.prettyhexrep(dest)}."
+
+def register(sideband=None):
+ registry.register(Command(
+ path=("conv",),
+ help_text="List all conversations",
+ help_detail="""conv
+ Lists all conversations.""",
+ max_args=0,
+ handler=_cmd_conv,
+ ))
+
+ registry.register(Command(
+ path=("conv", "send"),
+ help_text="Send an LXMF message to a destination",
+ help_detail="""conv send <dest> <message>
+ Sends an LXMF message to the specified destination hash.
+
+ <dest> Destination hash (32 hex chars) or known peer name
+ <message> Message content (rest of the line, quoted if needed)""",
+ min_args=2,
+ handler=_cmd_conv_send,
+ ))
+
+ registry.register(Command(
+ path=("conv", "command"),
+ help_text="Send an LXMF command to a destination",
+ help_detail="""conv command <dest> <command>
+ Sends an LXMF command message.
+
+ <dest> Destination hash (32 hex chars) or known peer name
+ <command> Command string (rest of the line)""",
+ min_args=2,
+ handler=_cmd_conv_command,
+ ))
+
+ registry.register(Command(
+ path=("conv", "announce"),
+ help_text="Send an LXMF announce",
+ help_detail="""conv announce
+ Announces the LXMF destination.""",
+ max_args=0,
+ handler=_cmd_conv_announce,
+ ))
+
+ registry.register(Command(
+ path=("conv", "sync"),
+ help_text="Manage LXMF propagation sync",
+ help_detail="""conv sync [limit]
+ Request message sync from the configured propagation node.
+
+conv sync cancel
+ Cancel an in-progress propagation sync.
+
+conv sync status
+ Show the current sync status and progress.""",
+ max_args=1,
+ handler=_cmd_conv_sync,
+ ))
+
+ registry.register(Command(
+ path=("conv", "sync", "cancel"),
+ help_text="Cancel in-progress LXMF sync",
+ max_args=0,
+ handler=_cmd_conv_sync_cancel,
+ ))
+
+ registry.register(Command(
+ path=("conv", "sync", "status"),
+ help_text="Show LXMF sync status",
+ max_args=0,
+ handler=_cmd_conv_sync_status,
+ ))
+
+ registry.register(Command(
+ path=("conv", "msgs"),
+ help_text="List messages in a conversation",
+ help_detail="""conv msgs <dest> [count]
+ Lists recent messages in the conversation with <dest>.
+ Default count is 12.
+
+ <dest> Destination hash (32 hex chars) or known peer name
+ [count] Maximum number of messages to show""",
+ min_args=1,
+ max_args=2,
+ handler=_cmd_conv_msgs,
+ ))
+
+ registry.register(Command(
+ path=("conv", "create"),
+ help_text="Create a new conversation",
+ help_detail="""conv create <dest> [name]
+ Creates a new conversation entry for the given destination.
+ Optionally sets a display name.
+
+ <dest> Destination hash (32 hex chars)
+ [name] Display name for the conversation""",
+ min_args=1,
+ max_args=2,
+ handler=_cmd_conv_create,
+ ))
+
+ registry.register(Command(
+ path=("conv", "delete"),
+ help_text="Delete a conversation and all its messages",
+ help_detail="""conv delete <dest>
+ Permanently deletes the conversation with <dest>, including all
+ messages and telemetry data.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_conv_delete,
+ ))
+
+ registry.register(Command(
+ path=("conv", "clear"),
+ help_text="Clear all messages in a conversation",
+ help_detail="""conv clear <dest>
+ Removes all messages from the conversation with <dest>,
+ but keeps the conversation entry itself.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_conv_clear,
+ ))
+
+ registry.register(Command(
+ path=("conv", "trust"),
+ help_text="Mark a conversation as trusted",
+ help_detail="""conv trust <dest>
+ Sets the trust flag on the conversation with <dest>.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_conv_trust,
+ ))
+
+ registry.register(Command(
+ path=("conv", "untrust"),
+ help_text="Mark a conversation as untrusted",
+ help_detail="""conv untrust <dest>
+ Removes the trust flag from the conversation with <dest>.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_conv_untrust,
+ ))
+
+ registry.register(Command(
+ path=("conv", "name"),
+ help_text="Set the display name for a conversation",
+ help_detail="""conv name <dest> <name>
+ Sets or updates the display name for the conversation with <dest>.
+
+ <dest> Destination hash (32 hex chars) or known peer name
+ <name> New display name""",
+ min_args=2,
+ handler=_cmd_conv_name,
+ ))
+
+ registry.register(Command(
+ path=("conv", "paper"),
+ help_text="Create a paper (offline) message",
+ help_detail="""conv paper <dest> <message>
+ Creates a paper message for offline delivery.
+
+ <dest> Destination hash (32 hex chars) or known peer name
+ <message> Message content""",
+ min_args=2,
+ handler=_cmd_conv_paper,
+ ))
diff --git a/sbapp/sideband/cli/cmd_net.py b/sbapp/sideband/cli/cmd_net.py
new file mode 100644
index 00000000..35bfb9aa
--- /dev/null
+++ b/sbapp/sideband/cli/cmd_net.py
@@ -0,0 +1,179 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import RNS
+from .commands import Command, registry
+from . import output as out
+
+
+def _cmd_net_propagation(args, sideband, ctx):
+ """Show or set the LXMF propagation node."""
+ if not args:
+ lines = []
+ lines.append(out.section("LXMF Propagation Node"))
+ configured = sideband.config.get("lxmf_propagation_node")
+ active = sideband.active_propagation_node
+
+ if configured: lines.append(out.kv("Configured", RNS.prettyhexrep(configured)))
+ else: lines.append(out.kv("Configured", "none"))
+
+ if active: lines.append(out.kv("Active", RNS.prettyhexrep(active)))
+ else: lines.append(out.kv("Active", "none"))
+
+ return "\n".join(lines)
+
+ return "Use 'net propagation set <hash>' to change the propagation node."
+
+
+def _cmd_net_propagation_set(args, sideband, ctx):
+ val = args[0].strip().lower()
+ if val in ("none", "null", ""):
+ sideband.set_active_propagation_node(None)
+ return "Propagation node cleared."
+
+ try:
+ dest = bytes.fromhex(val)
+ if len(dest) != RNS.Reticulum.TRUNCATED_HASHLENGTH // 8:
+ return f"Invalid destination hash: must be {RNS.Reticulum.TRUNCATED_HASHLENGTH // 8 * 2} hex characters."
+ except ValueError: return f"Invalid destination hash: {args[0]!r}"
+
+ sideband.set_active_propagation_node(dest)
+ return f"Propagation node set to {RNS.prettyhexrep(dest)}."
+
+def _cmd_net_announces(args, sideband, ctx):
+ announces = sideband.list_announces()
+ if not announces:
+ return "No announces in database."
+
+ lines = []
+ lines.append(out.section("Recent Announces"))
+ lines.append(f" {'Name':<20s} {'Type':<16s} {'Hash':<34s} {'Hops':>5s} {'Age':>10s}")
+ lines.append(f" {'-'*20} {'-'*16} {'-'*34} {'-'*5} {'-'*10}")
+
+ for ann in announces[:32]:
+ name = out.sanitize_name(ann.get("name", "-")) or "-"
+ name = name[:18]
+ dest = ann.get("dest")
+ hex_str = RNS.prettyhexrep(dest) if dest else "-"
+ atype = ann.get("type", "-") or "-"
+ hops = RNS.Transport.hops_to(dest) if dest else "?"
+ if hops == RNS.Transport.PATHFINDER_M: hops = "?"
+ age = out.time_ago(ann.get("time"))
+ lines.append(f" {name:<20s} {atype:<16s} {hex_str:<24s} {str(hops):>5s} {age:>10s}")
+
+ return "\n".join(lines)
+
+
+def _cmd_net_identity(args, sideband, ctx):
+ lines = []
+ lines.append(out.section("Identity"))
+
+ if sideband.identity: lines.append(out.kv("Identity hash", RNS.prettyhexrep(sideband.identity.hash)))
+ else: lines.append(out.kv("Identity hash", "-"))
+
+ if hasattr(sideband, "lxmf_destination") and sideband.lxmf_destination:
+ lines.append(out.kv("LXMF destination", RNS.prettyhexrep(sideband.lxmf_destination.hash)))
+ else:
+ lines.append(out.kv("LXMF destination", "-"))
+
+ lines.append(out.kv("Display name", sideband.config.get("display_name", "-")))
+
+ return "\n".join(lines)
+
+# TODO: Fix this so it also gets inbound links
+def _cmd_net_link(args, sideband, ctx):
+ dest_str = args[0].strip()
+ try: dest_hash = bytes.fromhex(dest_str)
+ except ValueError: return f"Invalid destination hash: {dest_str!r}"
+
+ link = None
+ try:
+ for l in RNS.Transport.active_links:
+ if hasattr(l, "destination") and l.destination and l.destination.hash == dest_hash:
+ link = l
+ break
+ except Exception: pass
+
+ lines = []
+ lines.append(out.section(f"Link with {RNS.prettyhexrep(dest_hash)}"))
+
+ if link is None:
+ lines.append("No active link to this destination.")
+ return "\n".join(lines)
+
+ try: lines.append(out.kv("Established", "yes" if link.status == RNS.Link.ACTIVE else "no"))
+ except Exception: pass
+
+ try: lines.append(out.kv("MTU", str(link.get_mtu())))
+ except Exception: pass
+
+ try:
+ rate = link.get_establishment_rate()
+ if rate: lines.append(out.kv("Establishment rate", f"{RNS.prettyspeed(rate)}"))
+ except Exception: pass
+
+ try:
+ erate = link.get_expected_rate()
+ if erate: lines.append(out.kv("Expected rate", f"{RNS.prettyspeed(erate)} bps"))
+ except Exception: pass
+
+ return "\n".join(lines)
+
+def register(sideband=None):
+ registry.register(Command(
+ path=("net", "propagation"),
+ help_text="Show or set the LXMF propagation node",
+ help_detail="""net propagation
+ Shows the currently configured and active propagation node.
+
+net propagation set <hash>
+ Sets the LXMF propagation node to the specified destination hash.
+ Pass 'none' to clear the propagation node.
+
+ <hash> Destination hash (32 hex chars) or 'none'""",
+ max_args=1,
+ handler=_cmd_net_propagation,
+ ))
+
+ registry.register(Command(
+ path=("net", "propagation", "set"),
+ help_text="Set the LXMF propagation node",
+ help_detail="""net propagation set <hash>
+ Sets the LXMF propagation node to the specified destination hash.
+ Use 'none' to clear the propagation node.
+
+ <hash> Destination hash (32 hex chars) or 'none'""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_net_propagation_set,
+ ))
+
+ registry.register(Command(
+ path=("net", "announces"),
+ help_text="List recent announces",
+ help_detail="""net announces
+ Lists recently received announces from the announce database.""",
+ max_args=0,
+ handler=_cmd_net_announces,
+ ))
+
+ registry.register(Command(
+ path=("net", "identity"),
+ help_text="Show own identity and destination information",
+ help_detail="""net identity
+ Displays the local Reticulum identity hash, the LXMF delivery
+ destination hash, and the configured display name.""",
+ max_args=0,
+ handler=_cmd_net_identity,
+ ))
+
+ registry.register(Command(
+ path=("net", "link"),
+ help_text="Show link statistics for a destination",
+ help_detail="""net link <dest>
+ Shows link-level statistics for the specified destination.
+
+ <dest> Destination hash (32 hex chars)""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_net_link,
+ ))
diff --git a/sbapp/sideband/cli/cmd_plugins.py b/sbapp/sideband/cli/cmd_plugins.py
new file mode 100644
index 00000000..132ae695
--- /dev/null
+++ b/sbapp/sideband/cli/cmd_plugins.py
@@ -0,0 +1,64 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+from .commands import Command, registry
+from . import output as out
+
+def _cmd_plugins(args, sideband, ctx):
+ lines = []
+ lines.append(out.section("Loaded Plugins"))
+
+ svc = list(sideband.active_service_plugins.values())
+ tel = list(sideband.active_telemetry_plugins.values())
+ cmd = list(sideband.active_command_plugins.values())
+
+ if svc:
+ lines.append("\nService plugins:")
+ for p in svc:
+ status = "running" if p.is_running() else "stopped"
+ lines.append(f" {p.service_name:<30s} {status}")
+
+ if tel:
+ lines.append("\nTelemetry plugins:")
+ for p in tel:
+ status = "running" if p.is_running() else "stopped"
+ lines.append(f" {p.plugin_name:<30s} {status}")
+
+ if cmd:
+ lines.append("\nCommand plugins:")
+ for p in cmd:
+ status = "running" if p.is_running() else "stopped"
+ lines.append(f" {p.command_name:<30s} {status}")
+
+ if not svc and not tel and not cmd: lines.append("No plugins currently loaded.")
+
+ return "\n".join(lines)+"\n"
+
+def _cmd_plugins_reload(args, sideband, ctx):
+ try:
+ sideband.reload_plugins()
+ svc = len(sideband.active_service_plugins)
+ tel = len(sideband.active_telemetry_plugins)
+ cmd = len(sideband.active_command_plugins)
+ total = svc + tel + cmd
+ return f"Plugins reloaded. {total} active ({svc} service, {tel} telemetry, {cmd} command)."
+ except Exception as e: return f"Error reloading plugins: {e}"
+
+def register(sideband=None):
+ registry.register(Command(
+ path=("plugins",),
+ help_text="List loaded plugins",
+ help_detail="""plugins
+ Lists all currently loaded and active plugins, grouped by type.
+ Each entry shows the plugin name and whether it is running.""",
+ max_args=0,
+ handler=_cmd_plugins,
+ ))
+
+ registry.register(Command(
+ path=("plugins", "reload"),
+ help_text="Reload plugins from the configured path",
+ help_detail="""plugins reload
+ Reloads all plugins from the configured plugin directory.""",
+ max_args=0,
+ handler=_cmd_plugins_reload,
+ ))
diff --git a/sbapp/sideband/cli/cmd_system.py b/sbapp/sideband/cli/cmd_system.py
new file mode 100644
index 00000000..2bad26ca
--- /dev/null
+++ b/sbapp/sideband/cli/cmd_system.py
@@ -0,0 +1,189 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import RNS
+from .commands import Command, registry, resolve_command, format_help, format_command_help
+from . import output as out
+
+def _cmd_help(args, sideband, ctx):
+ if not args:
+ return format_help(registry)
+
+ cmd, remaining = resolve_command(args, registry)
+ if cmd is not None:
+ return format_command_help(cmd)
+
+ prefix = tuple(args)
+ subcmds = registry.commands_at_level(prefix)
+ if subcmds:
+ lines = []
+ lines.append(f"Subcommands for {' '.join(prefix)}:\n")
+ for sc in sorted(subcmds, key=lambda c: c.name()):
+ lines.append(f" {sc.name():<30s} {sc.help_text}")
+ lines.append("")
+ lines.append(f"Use 'help {' '.join(prefix)} <subcommand>' for detailed help.")
+ return "\n".join(lines)
+
+ return f"Unknown command: {' '.join(args)}\nType 'help' for a list of available commands."
+
+def _cmd_raw(args, sideband, ctx):
+ expr = args[0] if args else ""
+ if not expr: return "Usage: raw <expression>\nExample: raw sideband.config[\"display_name\"]"
+
+ try:
+ result = eval(expr)
+ if result is not None: return repr(result)
+ return "None"
+ except Exception as e: return f"Error: {e}"
+
+def _cmd_log(args, sideband, ctx):
+ if not args:
+ level_names = {0: "squelch", 1: "critical", 2: "error", 3: "warning",
+ 4: "notice", 5: "info", 6: "debug", 7: "extreme"}
+ name = level_names.get(RNS.loglevel, str(RNS.loglevel))
+ return f"Current log level: {RNS.loglevel} ({name})"
+
+ try:
+ level = int(args[0])
+ if level < 0 or level > 7: return f"Invalid log level: {level}. Must be 0-7."
+ RNS.loglevel = level
+ level_names = {0: "squelch", 1: "critical", 2: "error", 3: "warning",
+ 4: "notice", 5: "info", 6: "debug", 7: "extreme"}
+ name = level_names.get(level, str(level))
+ return f"Log level set to {level} ({name})"
+
+ except ValueError: return f"Invalid log level: {args[0]}. Must be an integer 0-7."
+
+def _cmd_log_stream(args, sideband, ctx):
+ ctx.enter_stream_mode()
+ return None
+
+def _cmd_clear(args, sideband, ctx):
+ ctx.clear()
+ return None
+
+def _cmd_quit(args, sideband, ctx):
+ ctx.quit()
+ return None
+
+def _cmd_status(args, sideband, ctx):
+ lines = []
+ lines.append(out.section("Sideband Status"))# LXMF
+ if sideband.message_router: lines.append(out.kv("LXMF router", "active"))
+ else: lines.append(out.kv("LXMF router", "inactive"))
+
+ # Identity
+ lines.append(out.kv("Display name", sideband.config.get("display_name", "-")))
+ if sideband.identity: lines.append(out.kv("Identity hash", RNS.prettyhexrep(sideband.identity.hash)))
+ if hasattr(sideband, "lxmf_destination") and sideband.lxmf_destination: lines.append(out.kv("LXMF destination", RNS.prettyhexrep(sideband.lxmf_destination.hash)))
+
+ # LXMF
+ if sideband.message_router:
+ pn = sideband.message_router.get_outbound_propagation_node()
+ if pn: lines.append(out.kv("Propagation node", RNS.prettyhexrep(pn)))
+ else: lines.append(out.kv("Propagation node", "none"))
+
+ # Mode
+ if sideband.is_daemon: lines.append(out.kv("Mode", "daemon"))
+ elif sideband.is_standalone: lines.append(out.kv("Mode", "standalone"))
+ elif sideband.is_service: lines.append(out.kv("Mode", "service"))
+ elif sideband.is_client: lines.append(out.kv("Mode", "client"))
+
+ # Reticulum
+ if sideband.reticulum:
+ lines.append(out.kv("Reticulum", "running"))
+ lines.append(out.kv("Transport", "enabled" if RNS.Reticulum.transport_enabled() else "disabled"))
+ else: lines.append(out.kv("Reticulum", "not started"))
+
+ # Telemetry
+ lines.append(out.kv("Telemetry", out.bool_str(sideband.config.get("telemetry_enabled", False))))
+
+ # Voice
+ lines.append(out.kv("Voice", out.bool_str(sideband.config.get("voice_enabled", False))))
+ if hasattr(sideband, "voice_running"):
+ lines.append(out.kv("Voice running", out.bool_str(sideband.voice_running)))
+
+ # Conversations and announces
+ try:
+ convs = sideband.list_conversations()
+ lines.append(out.kv("Conversations", str(len(convs))))
+ except Exception: lines.append(out.kv("Conversations", "-"))
+
+ try:
+ announces = sideband.list_announces()
+ lines.append(out.kv("Announce Stream", str(len(announces))))
+ except Exception: lines.append(out.kv("Announce Stream", "-"))
+
+ # Plugins
+ try:
+ n_cmd = len(sideband.active_command_plugins)
+ n_svc = len(sideband.active_service_plugins)
+ n_tel = len(sideband.active_telemetry_plugins)
+ lines.append(out.kv("Plugins", f"cmd={n_cmd} svc={n_svc} telemetry={n_tel}"))
+ except Exception: lines.append(out.kv("Plugins", "-"))
+
+ return "\n".join(lines)
+
+def register(sideband=None):
+ registry.register(Command(
+ path=("help",),
+ help_text="Show help for all commands or a specific command",
+ help_detail="""help [command path]
+ With no arguments, lists all available commands.
+ With a command path, shows detailed help for that command.""",
+ max_args=10,
+ handler=_cmd_help,
+ ))
+
+ registry.register(Command(
+ path=("raw",),
+ help_text="Evaluate a Python expression on the sideband core",
+ help_detail="""raw <expression>
+ Evaluates the given Python expression with the Sideband core
+ bound as the global variable 'sideband'.
+ Example: raw sideband.config["display_name"]""",
+ min_args=1,
+ handler=_cmd_raw,
+ ))
+
+ registry.register(Command(
+ path=("log",),
+ help_text="Get or set the RNS log level",
+ help_detail="""log [level]
+ With no arguments, shows the current log level.
+ With a numeric argument (0-7), sets the log level.
+ 0=squelch 1=critical 2=error 3=warning 4=notice 5=info 6=debug 7=extreme""",
+ max_args=1,
+ handler=_cmd_log,
+ ))
+
+ registry.register(Command(
+ path=("log", "stream"),
+ help_text="Stream log output until Enter is pressed",
+ help_detail="""log stream
+ Enters log-streaming mode. All RNS log output is printed
+ directly to the terminal in real time. Press Enter to return
+ to the interactive prompt.""",
+ max_args=0,
+ handler=_cmd_log_stream,
+ ))
+
+ registry.register(Command(
+ path=("clear",),
+ help_text="Clear the console output",
+ handler=_cmd_clear,
+ ))
+
+ registry.register(Command(
+ path=("quit",),
+ help_text="Persist data and exit the console",
+ handler=_cmd_quit,
+ ))
+
+ registry.register(Command(
+ path=("status",),
+ help_text="Show overall Sideband status summary",
+ help_detail="""status
+ Displays a summary of the Sideband instance status""",
+ max_args=0,
+ handler=_cmd_status,
+ ))
\ No newline at end of file
diff --git a/sbapp/sideband/cli/cmd_telemetry.py b/sbapp/sideband/cli/cmd_telemetry.py
new file mode 100644
index 00000000..0a13af45
--- /dev/null
+++ b/sbapp/sideband/cli/cmd_telemetry.py
@@ -0,0 +1,352 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import RNS
+from .commands import Command, registry
+from . import output as out
+
+
+def _resolve_dest_hex(sideband, dest_str: str):
+ """Resolve a destination string to bytes, searching by hex or name."""
+ dest_str = dest_str.strip()
+ try:
+ dest = bytes.fromhex(dest_str)
+ if len(dest) == RNS.Reticulum.TRUNCATED_HASHLENGTH // 8:
+ return dest
+ except ValueError:
+ pass
+
+ try:
+ for conv in sideband.list_conversations():
+ name = sideband.peer_display_name(conv["dest"])
+ if name and name.lower() == dest_str.lower():
+ return conv["dest"]
+ except Exception:
+ pass
+
+ try:
+ for ann in sideband.list_announces():
+ name = ann.get("name", "")
+ if name and name.lower() == dest_str.lower():
+ return ann["dest"]
+ except Exception:
+ pass
+
+ return None
+
+
+def _format_telemetry_dict(telemetry: dict) -> list[str]:
+ """Format a telemetry dict (from Telemeter.read_all()) into lines."""
+ lines = []
+ if not telemetry:
+ lines.append(" No telemetry data available.")
+ return lines
+
+ for sensor, data in sorted(telemetry.items()):
+ if data is None:
+ continue
+ if isinstance(data, dict):
+ lines.append(f" {sensor}:")
+ for key, val in sorted(data.items()):
+ lines.append(f" {key}: {val}")
+ else:
+ lines.append(f" {sensor}: {data}")
+
+ return lines
+
+
+def _cmd_telemetry(args, sideband, ctx):
+ """Show own current telemetry or peer telemetry."""
+ if not args:
+ # Show own telemetry
+ if not sideband.config.get("telemetry_enabled", False):
+ return "Telemetry collection is disabled. Use 'telemetry on' to enable."
+
+ telemetry = sideband.get_telemetry()
+ lines = []
+ lines.append(out.section("Own Telemetry"))
+ lines.extend(_format_telemetry_dict(telemetry))
+ return "\n".join(lines)
+
+ # Show peer telemetry
+ dest_str = args[0]
+ dest = _resolve_dest_hex(sideband, dest_str)
+ if dest is None:
+ return f"Unknown destination: {dest_str}"
+
+ entries = sideband.peer_telemetry(dest, limit=1)
+ if not entries:
+ return f"No telemetry available for {RNS.prettyhexrep(dest)}."
+
+ from sbapp.sideband.sense import Telemeter
+ latest = entries[-1]
+ packed = latest[1]
+ telemeter = Telemeter.from_packed(packed)
+
+ lines = []
+ lines.append(out.section(f"Telemetry: {RNS.prettyhexrep(dest)}"))
+ if telemeter:
+ lines.extend(_format_telemetry_dict(telemeter.read_all()))
+ else:
+ lines.append(" Could not decode telemetry data.")
+
+ return "\n".join(lines)
+
+
+def _cmd_telemetry_request(args, sideband, ctx):
+ """Request telemetry from a peer."""
+ dest_str = args[0]
+ dest = _resolve_dest_hex(sideband, dest_str)
+ if dest is None:
+ return f"Unknown destination: {dest_str}"
+
+ result = sideband.request_latest_telemetry(from_addr=dest)
+ status_map = {
+ "sent": f"Telemetry request sent to {RNS.prettyhexrep(dest)}.",
+ "in_progress": "A telemetry request is already in progress for this peer.",
+ "destination_unknown": "Destination identity unknown. Path request initiated.",
+ "no_address": "Cannot request telemetry from self.",
+ "not_sent": "Failed to send telemetry request.",
+ }
+ return status_map.get(result, f"Request status: {result}")
+
+
+def _cmd_telemetry_send(args, sideband, ctx):
+ """Send own telemetry to a peer."""
+ dest_str = args[0]
+ dest = _resolve_dest_hex(sideband, dest_str)
+ if dest is None:
+ return f"Unknown destination: {dest_str}"
+
+ result = sideband.send_latest_telemetry(to_addr=dest)
+ if result:
+ return f"Telemetry sent to {RNS.prettyhexrep(dest)}."
+ else:
+ return f"Failed to send telemetry to {RNS.prettyhexrep(dest)}."
+
+
+def _cmd_telemetry_track(args, sideband, ctx):
+ """Start live tracking of a peer."""
+ dest_str = args[0]
+ try:
+ interval = int(args[1])
+ duration = int(args[2])
+ except ValueError:
+ return "Interval and duration must be integers (seconds)."
+
+ dest = _resolve_dest_hex(sideband, dest_str)
+ if dest is None:
+ return f"Unknown destination: {dest_str}"
+
+ sideband.start_tracking(dest, interval, duration)
+ return f"Tracking {RNS.prettyhexrep(dest)} every {interval}s for {duration}s."
+
+
+def _cmd_telemetry_untrack(args, sideband, ctx):
+ """Stop live tracking of a peer."""
+ dest_str = args[0]
+ dest = _resolve_dest_hex(sideband, dest_str)
+ if dest is None:
+ return f"Unknown destination: {dest_str}"
+
+ sideband.stop_tracking(dest)
+ return f"Stopped tracking {RNS.prettyhexrep(dest)}."
+
+
+def _cmd_telemetry_list(args, sideband, ctx):
+ """List telemetry entries from the database."""
+ dest = None
+ count = 20
+
+ if args:
+ # First arg could be dest or count
+ maybe_dest = _resolve_dest_hex(sideband, args[0])
+ if maybe_dest is not None:
+ dest = maybe_dest
+ if len(args) > 1:
+ try:
+ count = int(args[1])
+ except ValueError:
+ return f"Invalid count: {args[1]}"
+ else:
+ try:
+ count = int(args[0])
+ except ValueError:
+ return f"Unknown destination or invalid count: {args[0]}"
+
+ results = sideband.list_telemetry(context_dest=dest, limit=count)
+ if not results:
+ return "No telemetry entries found."
+
+ lines = []
+ lines.append(out.section("Telemetry Entries"))
+
+ from sbapp.sideband.sense import Telemeter
+
+ for source, entries in sorted(results.items(), key=lambda x: x[0].hex()):
+ name = sideband.peer_display_name(source)
+ lines.append(f"\n {name or RNS.prettyhexrep(source)}:")
+ for ts, packed in entries[:count]:
+ telemeter = Telemeter.from_packed(packed)
+ time_str = out.timestamp(ts)
+ if telemeter:
+ readings = telemeter.read_all()
+ sensors = ", ".join(sorted(readings.keys()))
+ lines.append(f" {time_str} ({sensors})")
+ else:
+ lines.append(f" {time_str} (undecodable)")
+
+ return "\n".join(lines)
+
+
+def _cmd_telemetry_on(args, sideband, ctx):
+ """Enable telemetry collection."""
+ sideband.config["telemetry_enabled"] = True
+ sideband.run_telemetry()
+ return "Telemetry collection enabled."
+
+
+def _cmd_telemetry_off(args, sideband, ctx):
+ """Disable telemetry collection."""
+ sideband.config["telemetry_enabled"] = False
+ sideband.stop_telemetry()
+ return "Telemetry collection disabled."
+
+
+def _cmd_telemetry_sensors(args, sideband, ctx):
+ """List available telemetry sensors and their status."""
+ lines = []
+ lines.append(out.section("Telemetry Sensors"))
+
+ sensors = [
+ ("location", "Location"),
+ ("information", "Information text"),
+ ("battery", "Battery level"),
+ ("pressure", "Barometric pressure"),
+ ("temperature", "Temperature"),
+ ("humidity", "Humidity"),
+ ("magnetic_field", "Magnetic field"),
+ ("ambient_light", "Ambient light"),
+ ("gravity", "Gravity"),
+ ("angular_velocity", "Angular velocity"),
+ ("acceleration", "Acceleration"),
+ ("proximity", "Proximity"),
+ ("rns_transport", "RNS transport stats"),
+ ]
+
+ for key, label in sensors:
+ enabled = sideband.config.get(f"telemetry_s_{key}", False)
+ status = "enabled" if enabled else "disabled"
+ lines.append(f" {label:<24s} {status}")
+
+ return "\n".join(lines)
+
+
+def register(sideband=None):
+ registry.register(Command(
+ path=("telemetry",),
+ help_text="Show own current telemetry or peer telemetry",
+ help_detail="""telemetry
+ Displays the current local telemetry readings from all active sensors.
+
+telemetry <dest>
+ Displays the latest known telemetry for the specified peer.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ max_args=1,
+ handler=_cmd_telemetry,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "request"),
+ help_text="Request telemetry from a peer",
+ help_detail="""telemetry request <dest>
+ Sends a telemetry request to the specified peer. The peer must
+ have telemetry enabled and allow requests from you.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_telemetry_request,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "send"),
+ help_text="Send own telemetry to a peer",
+ help_detail="""telemetry send <dest>
+ Sends the latest local telemetry to the specified peer.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_telemetry_send,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "track"),
+ help_text="Start live tracking of a peer",
+ help_detail="""telemetry track <dest> <interval> <duration>
+ Begins periodically requesting telemetry from the specified peer.
+
+ <dest> Destination hash (32 hex chars) or known peer name
+ <interval> Request interval in seconds
+ <duration> Total tracking duration in seconds""",
+ min_args=3,
+ max_args=3,
+ handler=_cmd_telemetry_track,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "untrack"),
+ help_text="Stop live tracking of a peer",
+ help_detail="""telemetry untrack <dest>
+ Stops the active live tracking session for the specified peer.
+
+ <dest> Destination hash (32 hex chars) or known peer name""",
+ min_args=1,
+ max_args=1,
+ handler=_cmd_telemetry_untrack,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "list"),
+ help_text="List telemetry entries from the database",
+ help_detail="""telemetry list [dest] [count]
+ Lists telemetry entries from the database. With no arguments,
+ lists recent telemetry across all peers. With a destination,
+ lists telemetry for that peer only.
+
+ [dest] Destination hash (32 hex chars) or known peer name
+ [count] Maximum number of entries to show (default: 20)""",
+ max_args=2,
+ handler=_cmd_telemetry_list,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "on"),
+ help_text="Enable telemetry collection",
+ help_detail="""telemetry on
+ Enables telemetry collection. Sensors configured in the config
+ will start producing readings. The telemeter is updated immediately.""",
+ max_args=0,
+ handler=_cmd_telemetry_on,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "off"),
+ help_text="Disable telemetry collection",
+ help_detail="""telemetry off
+ Disables telemetry collection. All sensors are stopped and
+ cached telemetry is cleared from memory.""",
+ max_args=0,
+ handler=_cmd_telemetry_off,
+ ))
+
+ registry.register(Command(
+ path=("telemetry", "sensors"),
+ help_text="List available telemetry sensors and their status",
+ help_detail="""telemetry sensors
+ Lists all available telemetry sensors, showing whether each
+ sensor is currently enabled or disabled based on the configuration.""",
+ max_args=0,
+ handler=_cmd_telemetry_sensors,
+ ))
diff --git a/sbapp/sideband/cli/cmd_voice.py b/sbapp/sideband/cli/cmd_voice.py
new file mode 100644
index 00000000..2dfd17d4
--- /dev/null
+++ b/sbapp/sideband/cli/cmd_voice.py
@@ -0,0 +1,119 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import RNS
+from .commands import Command, registry
+from . import output as out
+
+def register(sideband=None):
+ registry.register(Command(
+ path=("voice", "status"),
+ help_text="Show voice/telephony status",
+ help_detail="""voice status
+ Displays the current state of the voice/telephony subsystem:
+ - Whether voice is enabled and running
+ - Telephone availability and call state
+ - Active call details (if any)
+ - Audio profile and mode
+ - Transmit/receive mute status""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "announce"),
+ help_text="Send an LXST telephony announce",
+ help_detail="""voice announce
+ Announces the LXST telephony destination on all active interfaces.
+ Requires voice to be enabled.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "dial"),
+ help_text="Initiate a voice call",
+ help_detail="""voice dial <dest>
+ Initiates a voice call to the specified destination.
+
+ <dest> Destination identity hash (32 hex chars) of the peer's
+ LXST telephony identity""",
+ min_args=1,
+ max_args=2,
+ ))
+
+ registry.register(Command(
+ path=("voice", "hangup"),
+ help_text="End the active voice call",
+ help_detail="""voice hangup
+ Ends any active or connecting voice call.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "answer"),
+ help_text="Answer an incoming voice call",
+ help_detail="""voice answer
+ Answers an incoming (ringing) voice call.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "log"),
+ help_text="Show the call log",
+ help_detail="""voice log
+ Displays the recent call log, including call direction,
+ peer identity, timestamp, and outcome.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "mute", "mic"),
+ help_text="Mute the microphone (transmit)",
+ help_detail="""voice mute mic
+ Mutes the microphone, preventing audio transmission to the peer.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "mute", "spk"),
+ help_text="Mute the speaker (receive)",
+ help_detail="""voice mute spk
+ Mutes the speaker, preventing audio reception from the peer.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "unmute", "mic"),
+ help_text="Unmute the microphone (transmit)",
+ help_detail="""voice unmute mic
+ Unmutes the microphone, resuming audio transmission.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "unmute", "spk"),
+ help_text="Unmute the speaker (receive)",
+ help_detail="""voice unmute spk
+ Unmutes the speaker, resuming audio reception.""",
+ max_args=0,
+ ))
+
+ registry.register(Command(
+ path=("voice", "profile"),
+ help_text="Switch audio profile",
+ help_detail="""voice profile <name>
+ Switches to the specified audio profile.
+
+ <name> Profile name""",
+ min_args=1,
+ max_args=1,
+ ))
+
+ registry.register(Command(
+ path=("voice", "mode"),
+ help_text="Switch audio mode",
+ help_detail="""voice mode <name>
+ Switches to the specified call mode.
+
+ <name> Mode name""",
+ min_args=1,
+ max_args=1,
+ ))
\ No newline at end of file
diff --git a/sbapp/sideband/cli/commands.py b/sbapp/sideband/cli/commands.py
new file mode 100644
index 00000000..99313353
--- /dev/null
+++ b/sbapp/sideband/cli/commands.py
@@ -0,0 +1,159 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import shlex
+import textwrap
+
+class Command:
+
+ def __init__(self, path, help_text="", help_detail=None, min_args=0, max_args=None, aliases=None, handler=None):
+ self.path = tuple(path)
+ self.aliases = [tuple(a) for a in (aliases or [])]
+ self.help_text = help_text
+ self.help_detail = help_detail
+ self.min_args = min_args
+ self.max_args = max_args
+ self.handler = handler
+
+ def execute(self, args, sideband, ctx):
+ if self.handler is not None: return self.handler(args, sideband, ctx)
+ return "Not yet implemented"
+
+ def name(self):
+ return " ".join(self.path)
+
+ def validate_args(self, args):
+ if len(args) < self.min_args: return f"Too few arguments. {self.usage()}"
+ if self.max_args is not None and len(args) > self.max_args: return f"Too many arguments. {self.usage()}"
+ return None
+
+ def usage(self):
+ if self.help_detail:
+ first_line = self.help_detail.strip().split("\n")[0]
+ return first_line
+ return f"{self.name()}"
+
+class CommandRegistry:
+ def __init__(self):
+ self._commands = {}
+ self._aliases = {}
+
+ def register(self, command):
+ self._commands[command.path] = command
+ for alias in command.aliases: self._aliases[alias] = command.path
+
+ def unregister(self, path):
+ path = tuple(path)
+ if path in self._commands: del self._commands[path]
+ to_remove = [a for a, p in self._aliases.items() if p == path]
+ for a in to_remove: del self._aliases[a]
+
+ def lookup(self, path):
+ path = tuple(path)
+ if path in self._aliases: path = self._aliases[path]
+ return self._commands.get(path)
+
+ def all_commands(self):
+ return list(self._commands.values())
+
+ def commands_at_level(self, prefix):
+ prefix = tuple(prefix)
+ result = []
+ for cmd in self._commands.values():
+ if len(cmd.path) == len(prefix) + 1 and cmd.path[:len(prefix)] == prefix:
+ result.append(cmd)
+ return result
+
+ def subcommands(self, prefix):
+ prefix = tuple(prefix)
+ result = []
+ for cmd in self._commands.values():
+ if len(cmd.path) > len(prefix) and cmd.path[:len(prefix)] == prefix:
+ result.append(cmd)
+ return result
+
+ def build_help_tree(self):
+ tree = {}
+ for cmd in self._commands.values():
+ node = tree
+ for i, part in enumerate(cmd.path):
+ if i == len(cmd.path) - 1:
+ # Leaf: this is the final part of the path
+ if part in node and isinstance(node[part], dict):
+ # Already has subcommands, add the command alongside
+ node[part]["__cmd__"] = cmd
+ else:
+ node[part] = cmd
+ else:
+ # Intermediate: ensure there's a dict to descend into
+ if part not in node:
+ node[part] = {}
+ elif isinstance(node[part], Command):
+ # Convert the existing command into a dict with __cmd__
+ existing = node[part]
+ node[part] = {"__cmd__": existing}
+ node = node[part]
+ return tree
+
+registry = CommandRegistry()
+
+def parse_input(line):
+ try: return shlex.split(line)
+ except ValueError as e:
+ # Unbalanced quotes or similar
+ return line.split()
+
+def resolve_command(tokens, reg=registry):
+ if not tokens: return None, []
+
+ # Try progressively shorter path prefixes
+ for length in range(min(len(tokens), 4), 0, -1):
+ path = tuple(tokens[:length])
+ cmd = reg.lookup(path)
+ if cmd is not None:
+ return cmd, tokens[length:]
+
+ return None, tokens
+
+
+def format_help(reg=registry):
+ lines = []
+ lines.append("Available commands:\n")
+
+ tree = reg.build_help_tree()
+ def render_node(node, indent=0):
+ pad = " " * indent
+ for key in sorted(node.keys()):
+ if key == "__cmd__": continue
+ value = node[key]
+ if isinstance(value, Command):
+ cmd = value
+ help_str = f"{pad}{cmd.name()}"
+ lines.append(help_str)
+ elif isinstance(value, dict):
+ # Check if this node also has a direct command
+ if "__cmd__" in value:
+ cmd = value["__cmd__"]
+ help_str = f"{pad}{cmd.name()}"
+ lines.append(help_str)
+ else: lines.append(f"{pad}{key}")
+ render_node(value, indent + 1)
+
+ render_node(tree)
+
+ lines.append("")
+ lines.append("Use 'help <command>' for detailed help on a specific command")
+
+ return "\n".join(lines)
+
+def format_command_help(cmd):
+ lines = []
+ lines.append(f"Command: {cmd.name()}")
+ lines.append(f" {cmd.help_text}")
+ if cmd.aliases:
+ lines.append(f" Aliases: {', '.join(' '.join(a) for a in cmd.aliases)}")
+ if cmd.help_detail:
+ lines.append("")
+ lines.append("Usage:")
+ lines.append(textwrap.indent(cmd.help_detail, " "))
+ lines.append("")
+ return "\n".join(lines)
\ No newline at end of file
diff --git a/sbapp/sideband/cli/completer.py b/sbapp/sideband/cli/completer.py
new file mode 100644
index 00000000..bad62f25
--- /dev/null
+++ b/sbapp/sideband/cli/completer.py
@@ -0,0 +1,171 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+from .config_tree import get_categories, get_aliases, get_key_type, T_BOOL
+
+class DestinationCompleter:
+ def __init__(self, sideband=None): self._sideband = sideband
+ def complete(self, prefix: str) -> list[str]:
+ results = []
+ if self._sideband is None: return results
+
+ try:
+ for conv in self._sideband.list_conversations():
+ dest = conv["dest"]
+ hex_str = dest.hex() if isinstance(dest, bytes) else str(dest)
+ name = self._sideband.peer_display_name(dest)
+ if hex_str.startswith(prefix) or (name and name.lower().startswith(prefix.lower())): results.append(hex_str)
+ except Exception: pass
+
+ try:
+ for ann in self._sideband.list_announces():
+ dest = ann["dest"]
+ hex_str = dest.hex() if isinstance(dest, bytes) else str(dest)
+ name = ann.get("name", "")
+ if hex_str.startswith(prefix):
+ if hex_str not in results: results.append(hex_str)
+ except Exception: pass
+
+ return results
+
+class ConfigCategoryCompleter:
+ def complete(self, prefix: str) -> list[str]: return [c for c in get_categories() if c.startswith(prefix)]
+
+class AliasCompleter:
+ def __init__(self, category: str): self._category = category
+ def complete(self, prefix: str) -> list[str]: return [a for a in get_aliases(self._category) if a.startswith(prefix)]
+
+class ConfigValueCompleter:
+ STRING_OPTIONS = { "connect_ifmode_local": ["full", "gateway", "access point", "roaming", "boundary"],
+ "connect_ifmode_tcp": ["full", "gateway", "access point", "roaming", "boundary"],
+ "connect_ifmode_i2p": ["full", "gateway", "access point", "roaming", "boundary"],
+ "connect_ifmode_rnode": ["full", "gateway", "access point", "roaming", "boundary"],
+ "connect_ifmode_modem": ["full", "gateway", "access point", "roaming", "boundary"],
+ "connect_ifmode_serial": ["full", "gateway", "access point", "roaming", "boundary"],
+ "connect_ifmode_bluetooth": ["full", "gateway", "access point", "roaming", "boundary"],
+ "connect_ifmode_weave": ["full", "gateway", "access point", "roaming", "boundary"],
+ "hw_rnode_modulation": ["LoRa", "FSK"],
+ "hw_modem_parity": ["none", "even", "odd"],
+ "hw_serial_parity": ["none", "even", "odd"] }
+
+ def __init__(self, sideband=None, key=None):
+ self._sideband = sideband
+ self._key = key
+
+ def complete(self, prefix: str) -> list[str]:
+ if self._key is None: return []
+ vtype = get_key_type(self._key)
+ if vtype == T_BOOL: return [v for v in ["true", "false"] if v.startswith(prefix)]
+ if self._key in self.STRING_OPTIONS: return [v for v in self.STRING_OPTIONS[self._key] if v.startswith(prefix)]
+ return []
+
+class TelemetrySensorCompleter:
+ SENSORS = ["location", "information", "battery", "pressure", "temperature",
+ "humidity", "magnetic_field", "ambient_light", "gravity",
+ "angular_velocity", "acceleration", "proximity", "rns_transport",
+ "power_consumption", "power_production", "processor", "ram", "nvm",
+ "custom", "tank", "fuel", "lxmf_propagation", "connection_map"]
+
+ def complete(self, prefix: str) -> list[str]: return [s for s in self.SENSORS if s.startswith(prefix)]
+
+def _build_alias_subtree():
+ subtree = {}
+ for cat in get_categories(): subtree[cat] = AliasCompleter(cat)
+ return subtree
+
+def _build_help_tree(command_tree):
+ if not isinstance(command_tree, dict): return None
+ result = {}
+ for key, value in command_tree.items():
+ if key == "help": continue
+ if isinstance(value, dict):
+ subtree = _build_help_tree(value)
+ result[key] = subtree if subtree is not None else None
+ else: result[key] = None
+ return result
+
+def build_completion_tree(sideband=None):
+ dest_completer = DestinationCompleter(sideband)
+ sensor_completer = TelemetrySensorCompleter()
+ config_alias_subtree = _build_alias_subtree()
+
+ tree = {
+ "raw": None,
+ "log": {
+ "stream": None,
+ },
+ "clear": None,
+ "quit": None,
+ "status": None,
+
+ "conv": {
+ "send": dest_completer,
+ "command": dest_completer,
+ "announce": None,
+ "sync": {
+ "cancel": None,
+ "status": None,
+ },
+ "msgs": dest_completer,
+ "create": dest_completer,
+ "delete": dest_completer,
+ "clear": dest_completer,
+ "trust": dest_completer,
+ "untrust": dest_completer,
+ "name": dest_completer,
+ "paper": dest_completer,
+ },
+
+ "config": {
+ "list": ConfigCategoryCompleter(),
+ "get": config_alias_subtree,
+ "set": config_alias_subtree,
+ "save": None,
+ "reload": None,
+ },
+
+ "telemetry": {
+ "request": dest_completer,
+ "send": dest_completer,
+ "track": dest_completer,
+ "untrack": dest_completer,
+ "list": dest_completer,
+ "on": None,
+ "off": None,
+ "sensors": None,
+ },
+
+ "voice": {
+ "status": None,
+ "announce": None,
+ "dial": dest_completer,
+ "hangup": None,
+ "answer": None,
+ "log": None,
+ "mute": {
+ "mic": None,
+ "spk": None,
+ },
+ "unmute": {
+ "mic": None,
+ "spk": None,
+ },
+ "profile": None,
+ "mode": None,
+ },
+
+ "net": {
+ "propagation": {
+ "set": None,
+ },
+ "announces": None,
+ "identity": None,
+ "link": dest_completer,
+ },
+
+ "plugins": {
+ "reload": None,
+ },
+ }
+
+ tree["help"] = _build_help_tree(tree)
+ return tree
diff --git a/sbapp/sideband/cli/config_tree.py b/sbapp/sideband/cli/config_tree.py
new file mode 100644
index 00000000..01a86bea
--- /dev/null
+++ b/sbapp/sideband/cli/config_tree.py
@@ -0,0 +1,283 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+T_BOOL = "bool"
+T_INT = "int"
+T_FLOAT = "float"
+T_STR = "str"
+T_LIST = "list"
+T_NONE = "None"
+
+CONFIG_METADATA = {
+ "debug": {"category": "general", "alias": "debug", "type": T_BOOL, "description": "Enable debug logging"},
+ "display_name": {"category": "general", "alias": "display_name", "type": T_STR, "description": "Display name shown in announces"},
+ "notifications_on": {"category": "general", "alias": "notifications_on", "type": T_BOOL, "description": "Enable OS notifications"},
+ "dark_ui": {"category": "general", "alias": "dark_ui", "type": T_BOOL, "description": "Use dark UI theme"},
+ "start_announce": {"category": "general", "alias": "start_announce", "type": T_BOOL, "description": "Automatically announce on startup"},
+ "start_at_boot": {"category": "general", "alias": "start_at_boot", "type": T_BOOL, "description": "Start Sideband at boot (Android)"},
+ "print_command": {"category": "general", "alias": "print_command", "type": T_STR, "description": "Print command for paper messages"},
+ "eink_mode": {"category": "general", "alias": "eink_mode", "type": T_BOOL, "description": "Optimise display for e-ink"},
+ "lxm_limit_1mb": {"category": "general", "alias": "lxm_limit_1mb", "type": T_BOOL, "description": "Limit LXMF messages to 1MB transfer"},
+ "trusted_markup_only": {"category": "general", "alias": "trusted_markup_only", "type": T_BOOL, "description": "Only render markup from trusted peers"},
+ "compose_in_markdown": {"category": "general", "alias": "compose_in_markdown", "type": T_BOOL, "description": "Compose messages in Markdown by default"},
+ "confirm_calls": {"category": "general", "alias": "confirm_calls", "type": T_BOOL, "description": "Require confirmation before answering calls"},
+ "classic_message_colors": {"category": "general", "alias": "classic_message_colors", "type": T_BOOL, "description": "Use classic message color scheme"},
+ "display_style_in_contact_list": {"category": "general", "alias": "display_style_in_contact_list", "type": T_BOOL, "description": "Display peer appearance in contact list"},
+ "hq_ptt": {"category": "general", "alias": "hq_ptt", "type": T_BOOL, "description": "High-quality PTT audio"},
+ "input_language": {"category": "general", "alias": "input_language", "type": T_STR, "description": "Override input language"},
+ "allow_predictive_text": {"category": "general", "alias": "allow_predictive_text", "type": T_BOOL, "description": "Allow predictive text input"},
+ "block_predictive_text": {"category": "general", "alias": "block_predictive_text", "type": T_BOOL, "description": "Block predictive text input"},
+ "config_template": {"category": "general", "alias": "config_template", "type": T_STR, "description": "Custom RNS configuration template"},
+
+ "lxmf_propagation_node": {"category": "lxmf", "alias": "propagation_node", "type": T_STR, "description": "Configured propagation node hash"},
+ "lxmf_sync_limit": {"category": "lxmf", "alias": "sync_limit", "type": T_INT, "description": "Maximum messages per sync (None=unlimited)"},
+ "lxmf_sync_max": {"category": "lxmf", "alias": "sync_max", "type": T_INT, "description": "Maximum sync retries"},
+ "lxmf_periodic_sync": {"category": "lxmf", "alias": "periodic_sync", "type": T_BOOL, "description": "Enable periodic LXMF sync"},
+ "lxmf_ignore_unknown": {"category": "lxmf", "alias": "ignore_unknown", "type": T_BOOL, "description": "Drop messages from unknown senders"},
+ "lxmf_sync_interval": {"category": "lxmf", "alias": "sync_interval", "type": T_INT, "description": "Periodic sync interval in seconds"},
+ "lxmf_require_stamps": {"category": "lxmf", "alias": "require_stamps", "type": T_BOOL, "description": "Require stamps for inbound messages"},
+ "lxmf_inbound_stamp_cost": {"category": "lxmf", "alias": "inbound_stamp_cost", "type": T_INT, "description": "Inbound stamp cost (None=default)"},
+ "lxmf_try_propagation_on_fail": {"category": "lxmf", "alias": "try_propagation_on_fail", "type": T_BOOL, "description": "Fall back to propagation on delivery failure"},
+ "lxmf_ignore_invalid_stamps": {"category": "lxmf", "alias": "ignore_invalid_stamps", "type": T_BOOL, "description": "Ignore messages with invalid stamps"},
+ "last_lxmf_propagation_node": {"category": "lxmf", "alias": "last_propagation_node", "type": T_STR, "description": "Last used propagation node hash"},
+ "propagation_by_default": {"category": "lxmf", "alias": "propagation_by_default", "type": T_BOOL, "description": "Use propagation by default for new messages"},
+ "home_node_as_broadcast_repeater":{"category": "lxmf", "alias": "broadcast_repeater", "type": T_BOOL, "description": "Use home node as broadcast repeater"},
+ "send_telemetry_to_home_node": {"category": "lxmf", "alias": "send_telemetry_to_home_node","type": T_BOOL, "description": "Send telemetry to home node"},
+ "nn_home_node": {"category": "lxmf", "alias": "home_node", "type": T_STR, "description": "Home node hash"},
+
+ "connect_transport": {"category": "connect", "alias": "transport", "type": T_BOOL, "description": "Enable Reticulum Transport mode"},
+ "connect_local": {"category": "connect", "alias": "local", "type": T_BOOL, "description": "Enable AutoInterface (local network)"},
+ "connect_local_groupid": {"category": "connect", "alias": "local_groupid", "type": T_STR, "description": "AutoInterface group ID"},
+ "connect_local_ifac_netname": {"category": "connect", "alias": "local_ifac_netname", "type": T_STR, "description": "AutoInterface IFAC network name"},
+ "connect_local_ifac_passphrase": {"category": "connect", "alias": "local_ifac_passphrase", "type": T_STR, "description": "AutoInterface IFAC passphrase"},
+ "connect_tcp": {"category": "connect", "alias": "tcp", "type": T_BOOL, "description": "Enable TCP interface"},
+ "connect_tcp_host": {"category": "connect", "alias": "tcp_host", "type": T_STR, "description": "TCP interface host"},
+ "connect_tcp_port": {"category": "connect", "alias": "tcp_port", "type": T_STR, "description": "TCP interface port"},
+ "connect_tcp_ifac_netname": {"category": "connect", "alias": "tcp_ifac_netname", "type": T_STR, "description": "TCP IFAC network name"},
+ "connect_tcp_ifac_passphrase": {"category": "connect", "alias": "tcp_ifac_passphrase", "type": T_STR, "description": "TCP IFAC passphrase"},
+ "connect_i2p": {"category": "connect", "alias": "i2p", "type": T_BOOL, "description": "Enable I2P interface"},
+ "connect_i2p_b32": {"category": "connect", "alias": "i2p_b32", "type": T_STR, "description": "I2P B32 address"},
+ "connect_i2p_ifac_netname": {"category": "connect", "alias": "i2p_ifac_netname", "type": T_STR, "description": "I2P IFAC network name"},
+ "connect_i2p_ifac_passphrase": {"category": "connect", "alias": "i2p_ifac_passphrase", "type": T_STR, "description": "I2P IFAC passphrase"},
+ "connect_rnode": {"category": "connect", "alias": "rnode", "type": T_BOOL, "description": "Enable RNode interface"},
+ "connect_rnode_ifac_netname": {"category": "connect", "alias": "rnode_ifac_netname", "type": T_STR, "description": "RNode IFAC network name"},
+ "connect_rnode_ifac_passphrase": {"category": "connect", "alias": "rnode_ifac_passphrase", "type": T_STR, "description": "RNode IFAC passphrase"},
+ "connect_serial": {"category": "connect", "alias": "serial", "type": T_BOOL, "description": "Enable serial interface"},
+ "connect_serial_ifac_netname": {"category": "connect", "alias": "serial_ifac_netname", "type": T_STR, "description": "Serial IFAC network name"},
+ "connect_serial_ifac_passphrase": {"category": "connect", "alias": "serial_ifac_passphrase", "type": T_STR, "description": "Serial IFAC passphrase"},
+ "connect_modem": {"category": "connect", "alias": "modem", "type": T_BOOL, "description": "Enable modem interface"},
+ "connect_modem_ifac_netname": {"category": "connect", "alias": "modem_ifac_netname", "type": T_STR, "description": "Modem IFAC network name"},
+ "connect_modem_ifac_passphrase": {"category": "connect", "alias": "modem_ifac_passphrase", "type": T_STR, "description": "Modem IFAC passphrase"},
+ "connect_weave": {"category": "connect", "alias": "weave", "type": T_BOOL, "description": "Enable Weave interface"},
+ "connect_weave_ifac_netname": {"category": "connect", "alias": "weave_ifac_netname", "type": T_STR, "description": "Weave IFAC network name"},
+ "connect_weave_ifac_passphrase": {"category": "connect", "alias": "weave_ifac_passphrase", "type": T_STR, "description": "Weave IFAC passphrase"},
+ "connect_obfuscate_hops": {"category": "connect", "alias": "obfuscate_hops", "type": T_BOOL, "description": "Obfuscate hop counts"},
+ "connect_ifmode_local": {"category": "connect", "alias": "ifmode_local", "type": T_STR, "description": "AutoInterface mode (full/gateway/access point/roaming/boundary)"},
+ "connect_ifmode_tcp": {"category": "connect", "alias": "ifmode_tcp", "type": T_STR, "description": "TCP interface mode"},
+ "connect_ifmode_i2p": {"category": "connect", "alias": "ifmode_i2p", "type": T_STR, "description": "I2P interface mode"},
+ "connect_ifmode_rnode": {"category": "connect", "alias": "ifmode_rnode", "type": T_STR, "description": "RNode interface mode"},
+ "connect_ifmode_modem": {"category": "connect", "alias": "ifmode_modem", "type": T_STR, "description": "Modem interface mode"},
+ "connect_ifmode_serial": {"category": "connect", "alias": "ifmode_serial", "type": T_STR, "description": "Serial interface mode"},
+ "connect_ifmode_bluetooth": {"category": "connect", "alias": "ifmode_bluetooth", "type": T_STR, "description": "Bluetooth interface mode"},
+ "connect_ifmode_weave": {"category": "connect", "alias": "ifmode_weave", "type": T_STR, "description": "Weave interface mode"},
+ "connect_share_instance": {"category": "connect", "alias": "share_instance", "type": T_BOOL, "description": "Share Reticulum instance (Android)"},
+ "discover_interfaces": {"category": "connect", "alias": "discover_interfaces", "type": T_BOOL, "description": "Enable automatic interface discovery"},
+
+ "hw_rnode_frequency": {"category": "rnode", "alias": "frequency", "type": T_FLOAT, "description": "RNode frequency in Hz"},
+ "hw_rnode_modulation": {"category": "rnode", "alias": "modulation", "type": T_STR, "description": "RNode modulation (LoRa/FSK)"},
+ "hw_rnode_bandwidth": {"category": "rnode", "alias": "bandwidth", "type": T_INT, "description": "RNode bandwidth in Hz"},
+ "hw_rnode_spreading_factor": {"category": "rnode", "alias": "spreading_factor", "type": T_INT, "description": "RNode spreading factor"},
+ "hw_rnode_coding_rate": {"category": "rnode", "alias": "coding_rate", "type": T_INT, "description": "RNode coding rate (5-8)"},
+ "hw_rnode_tx_power": {"category": "rnode", "alias": "tx_power", "type": T_INT, "description": "RNode TX power in dBm (0=max)"},
+ "hw_rnode_beaconinterval": {"category": "rnode", "alias": "beacon_interval", "type": T_INT, "description": "RNode beacon interval in seconds"},
+ "hw_rnode_beacondata": {"category": "rnode", "alias": "beacon_data", "type": T_STR, "description": "RNode beacon data/callsign"},
+ "hw_rnode_bluetooth": {"category": "rnode", "alias": "bluetooth", "type": T_BOOL, "description": "Allow RNode Bluetooth"},
+ "hw_rnode_ble": {"category": "rnode", "alias": "ble", "type": T_BOOL, "description": "Force RNode BLE connection"},
+ "hw_rnode_bt_device": {"category": "rnode", "alias": "bt_device", "type": T_STR, "description": "RNode Bluetooth device name"},
+ "hw_rnode_tcp": {"category": "rnode", "alias": "tcp", "type": T_BOOL, "description": "Connect to RNode over TCP"},
+ "hw_rnode_tcp_host": {"category": "rnode", "alias": "tcp_host", "type": T_STR, "description": "RNode TCP host address"},
+ "hw_rnode_atl_short": {"category": "rnode", "alias": "atl_short", "type": T_FLOAT, "description": "RNode short airtime limit"},
+ "hw_rnode_atl_long": {"category": "rnode", "alias": "atl_long", "type": T_FLOAT, "description": "RNode long airtime limit"},
+ "hw_rnode_enable_framebuffer": {"category": "rnode", "alias": "enable_framebuffer", "type": T_BOOL, "description": "Enable RNode framebuffer display"},
+
+ "hw_modem_baudrate": {"category": "modem", "alias": "baudrate", "type": T_INT, "description": "Modem baud rate"},
+ "hw_modem_databits": {"category": "modem", "alias": "databits", "type": T_INT, "description": "Modem data bits"},
+ "hw_modem_stopbits": {"category": "modem", "alias": "stopbits", "type": T_INT, "description": "Modem stop bits"},
+ "hw_modem_parity": {"category": "modem", "alias": "parity", "type": T_STR, "description": "Modem parity (none/even/odd)"},
+ "hw_modem_preamble": {"category": "modem", "alias": "preamble", "type": T_INT, "description": "Modem preamble"},
+ "hw_modem_tail": {"category": "modem", "alias": "tail", "type": T_INT, "description": "Modem tail"},
+ "hw_modem_persistence": {"category": "modem", "alias": "persistence", "type": T_INT, "description": "Modem persistence"},
+ "hw_modem_slottime": {"category": "modem", "alias": "slottime", "type": T_INT, "description": "Modem slot time"},
+ "hw_modem_beaconinterval": {"category": "modem", "alias": "beacon_interval", "type": T_INT, "description": "Modem beacon interval"},
+ "hw_modem_beacondata": {"category": "modem", "alias": "beacon_data", "type": T_STR, "description": "Modem beacon data"},
+
+ "hw_serial_baudrate": {"category": "serial", "alias": "baudrate", "type": T_INT, "description": "Serial baud rate"},
+ "hw_serial_databits": {"category": "serial", "alias": "databits", "type": T_INT, "description": "Serial data bits"},
+ "hw_serial_stopbits": {"category": "serial", "alias": "stopbits", "type": T_INT, "description": "Serial stop bits"},
+ "hw_serial_parity": {"category": "serial", "alias": "parity", "type": T_STR, "description": "Serial parity (none/even/odd)"},
+
+ "telemetry_s_location": {"category": "sensors", "alias": "location", "type": T_BOOL, "description": "Sensor: location"},
+ "telemetry_s_battery": {"category": "sensors", "alias": "battery", "type": T_BOOL, "description": "Sensor: battery"},
+ "telemetry_s_pressure": {"category": "sensors", "alias": "pressure", "type": T_BOOL, "description": "Sensor: pressure"},
+ "telemetry_s_temperature": {"category": "sensors", "alias": "temperature", "type": T_BOOL, "description": "Sensor: temperature"},
+ "telemetry_s_humidity": {"category": "sensors", "alias": "humidity", "type": T_BOOL, "description": "Sensor: humidity"},
+ "telemetry_s_magnetic_field": {"category": "sensors", "alias": "magnetic_field", "type": T_BOOL, "description": "Sensor: magnetic field"},
+ "telemetry_s_ambient_light": {"category": "sensors", "alias": "ambient_light", "type": T_BOOL, "description": "Sensor: ambient light"},
+ "telemetry_s_gravity": {"category": "sensors", "alias": "gravity", "type": T_BOOL, "description": "Sensor: gravity"},
+ "telemetry_s_angular_velocity": {"category": "sensors", "alias": "angular_velocity", "type": T_BOOL, "description": "Sensor: angular velocity"},
+ "telemetry_s_acceleration": {"category": "sensors", "alias": "acceleration", "type": T_BOOL, "description": "Sensor: acceleration"},
+ "telemetry_s_proximity": {"category": "sensors", "alias": "proximity", "type": T_BOOL, "description": "Sensor: proximity"},
+ "telemetry_s_rns_transport": {"category": "sensors", "alias": "rns_transport", "type": T_BOOL, "description": "Sensor: RNS transport stats"},
+ "telemetry_s_fixed_location": {"category": "sensors", "alias": "fixed_location", "type": T_BOOL, "description": "Sensor: use fixed location"},
+ "telemetry_s_fixed_latlon": {"category": "sensors", "alias": "fixed_latlon", "type": T_LIST, "description": "Fixed location [lat, lon]"},
+ "telemetry_s_fixed_altitude": {"category": "sensors", "alias": "fixed_altitude", "type": T_FLOAT, "description": "Fixed altitude in meters"},
+ "telemetry_s_information": {"category": "sensors", "alias": "information", "type": T_BOOL, "description": "Sensor: information text"},
+ "telemetry_s_information_text": {"category": "sensors", "alias": "information_text", "type": T_STR, "description": "Information text content"},
+
+ "voice_enabled": {"category": "voice", "alias": "enabled", "type": T_BOOL, "description": "Enable voice/telephony"},
+ "voice_output": {"category": "voice", "alias": "output", "type": T_STR, "description": "Audio output device"},
+ "voice_input": {"category": "voice", "alias": "input", "type": T_STR, "description": "Audio input device"},
+ "voice_ringer": {"category": "voice", "alias": "ringer", "type": T_STR, "description": "Ringer audio device"},
+ "voice_trusted_only": {"category": "voice", "alias": "trusted_only", "type": T_BOOL, "description": "Allow calls from trusted peers only"},
+ "voice_low_latency": {"category": "voice", "alias": "low_latency", "type": T_BOOL, "description": "Enable low-latency audio output"},
+
+ "map_history_limit": {"category": "map", "alias": "history_limit", "type": T_INT, "description": "Map telemetry history limit in seconds"},
+ "map_lat": {"category": "map", "alias": "lat", "type": T_FLOAT, "description": "Map center latitude"},
+ "map_lon": {"category": "map", "alias": "lon", "type": T_FLOAT, "description": "Map center longitude"},
+ "map_zoom": {"category": "map", "alias": "zoom", "type": T_INT, "description": "Map zoom level"},
+ "map_storage_external": {"category": "map", "alias": "storage_external", "type": T_BOOL, "description": "Use external storage for map tiles"},
+ "map_use_offline": {"category": "map", "alias": "use_offline", "type": T_BOOL, "description": "Use offline map tiles"},
+ "map_use_online": {"category": "map", "alias": "use_online", "type": T_BOOL, "description": "Use online map tiles"},
+ "map_layer": {"category": "map", "alias": "layer", "type": T_STR, "description": "Map tile layer"},
+ "map_cluster": {"category": "map", "alias": "cluster", "type": T_BOOL, "description": "Cluster map markers"},
+ "map_interfaces": {"category": "map", "alias": "interfaces", "type": T_BOOL, "description": "Show interfaces on map"},
+ "map_storage_path": {"category": "map", "alias": "storage_path", "type": T_STR, "description": "Map tile storage path"},
+ "map_storage_file": {"category": "map", "alias": "storage_file", "type": T_STR, "description": "Map tile storage file"},
+
+ "service_plugins_enabled": {"category": "plugins", "alias": "services", "type": T_BOOL, "description": "Enable service plugins"},
+ "command_plugins_enabled": {"category": "plugins", "alias": "commands", "type": T_BOOL, "description": "Enable command plugins"},
+ "command_plugins_path": {"category": "plugins", "alias": "path", "type": T_STR, "description": "Path to plugin directory"},
+
+ "telemetry_enabled": {"category": "telemetry", "alias": "enabled", "type": T_BOOL, "description": "Enable telemetry collection"},
+ "telemetry_icon": {"category": "telemetry", "alias": "icon", "type": T_STR, "description": "Telemetry appearance icon name"},
+ "telemetry_fg": {"category": "telemetry", "alias": "fg", "type": T_LIST, "description": "Telemetry foreground color [r,g,b,a]"},
+ "telemetry_bg": {"category": "telemetry", "alias": "bg", "type": T_LIST, "description": "Telemetry background color [r,g,b,a]"},
+ "telemetry_send_to_trusted": {"category": "telemetry", "alias": "send_to_trusted", "type": T_BOOL, "description": "Send telemetry to trusted peers"},
+ "telemetry_send_to_collector": {"category": "telemetry", "alias": "send_to_collector", "type": T_BOOL, "description": "Send telemetry to collector"},
+ "telemetry_collector": {"category": "telemetry", "alias": "collector", "type": T_STR, "description": "Collector destination hash"},
+ "telemetry_request_from_collector": {"category": "telemetry", "alias": "request_from_collector", "type": T_BOOL, "description": "Request telemetry from collector"},
+ "telemetry_send_interval": {"category": "telemetry", "alias": "send_interval", "type": T_INT, "description": "Telemetry send interval in seconds"},
+ "telemetry_request_interval": {"category": "telemetry", "alias": "request_interval", "type": T_INT, "description": "Telemetry request interval in seconds"},
+ "telemetry_collector_enabled": {"category": "telemetry", "alias": "collector_enabled", "type": T_BOOL, "description": "Enable telemetry collector mode"},
+ "telemetry_to_mqtt": {"category": "telemetry", "alias": "to_mqtt", "type": T_BOOL, "description": "Forward telemetry to MQTT"},
+ "telemetry_mqtt_host": {"category": "telemetry", "alias": "mqtt_host", "type": T_STR, "description": "MQTT broker host"},
+ "telemetry_mqtt_port": {"category": "telemetry", "alias": "mqtt_port", "type": T_INT, "description": "MQTT broker port"},
+ "telemetry_mqtt_user": {"category": "telemetry", "alias": "mqtt_user", "type": T_STR, "description": "MQTT username"},
+ "telemetry_mqtt_pass": {"category": "telemetry", "alias": "mqtt_pass", "type": T_STR, "description": "MQTT password"},
+ "telemetry_mqtt_validate_ssl": {"category": "telemetry", "alias": "mqtt_validate_ssl", "type": T_BOOL, "description": "Validate MQTT SSL certificate"},
+ "telemetry_send_appearance": {"category": "telemetry", "alias": "send_appearance", "type": T_BOOL, "description": "Send appearance with telemetry"},
+ "telemetry_display_trusted_only": {"category": "telemetry", "alias": "display_trusted_only", "type": T_BOOL, "description": "Display telemetry from trusted only"},
+ "display_style_from_all": {"category": "telemetry", "alias": "display_style_from_all", "type": T_BOOL, "description": "Display appearance from all peers"},
+ "telemetry_receive_trusted_only": {"category": "telemetry", "alias": "receive_trusted_only", "type": T_BOOL, "description": "Receive telemetry from trusted only"},
+ "telemetry_send_all_to_collector": {"category": "telemetry", "alias": "send_all_to_collector", "type": T_BOOL, "description": "Send all telemetry to collector"},
+ "telemetry_use_propagation_only": {"category": "telemetry", "alias": "use_propagation_only", "type": T_BOOL, "description": "Send telemetry via propagation only"},
+ "telemetry_try_propagation_on_fail": {"category": "telemetry", "alias": "try_propagation_on_fail", "type": T_BOOL, "description": "Fall back to propagation for telemetry"},
+ "telemetry_requests_only_send_latest": {"category": "telemetry", "alias": "requests_only_send_latest", "type": T_BOOL, "description": "Only send latest telemetry on requests"},
+ "telemetry_allow_requests_from_trusted": {"category": "telemetry", "alias": "allow_requests_from_trusted", "type": T_BOOL, "description": "Allow telemetry requests from trusted"},
+ "telemetry_allow_requests_from_anyone": {"category": "telemetry", "alias": "allow_requests_from_anyone", "type": T_BOOL, "description": "Allow telemetry requests from anyone"},
+}
+
+def get_categories():
+ cats = set()
+ for meta in CONFIG_METADATA.values(): cats.add(meta["category"])
+ return sorted(cats)
+
+def get_keys_in_category(category):
+ keys = []
+ for key, meta in CONFIG_METADATA.items():
+ if meta["category"] == category: keys.append(key)
+ return sorted(keys)
+
+def get_aliases(category):
+ aliases = []
+ for meta in CONFIG_METADATA.values():
+ if meta["category"] == category: aliases.append(meta["alias"])
+ return sorted(aliases)
+
+def resolve_alias(category, alias):
+ for key, meta in CONFIG_METADATA.items():
+ if meta["category"] == category and meta["alias"] == alias: return key
+ return None
+
+def get_key_info(key):
+ return CONFIG_METADATA.get(key)
+
+def get_all_keys():
+ return sorted(CONFIG_METADATA.keys())
+
+def get_key_type(key):
+ meta = get_key_info(key)
+ if meta: return meta["type"]
+ return None
+
+def get_key_description(key):
+ meta = get_key_info(key)
+ if meta: return meta["description"]
+ return None
+
+def get_key_category(key):
+ meta = get_key_info(key)
+ if meta: return meta["category"]
+ return None
+
+def get_alias_for_key(key):
+ meta = get_key_info(key)
+ if meta: return meta["alias"]
+ return None
+
+def convert_value(key, raw_value):
+ vtype = get_key_type(key)
+ if vtype is None: return raw_value
+
+ if vtype == T_BOOL:
+ vl = raw_value.strip().lower()
+ if vl in ("true", "yes", "on", "1", "enable", "enabled"): return True
+ elif vl in ("false", "no", "off", "0", "disable", "disabled"): return False
+ else: raise ValueError(f"Invalid boolean value: {raw_value!r}. Use true/false, yes/no, on/off, or 1/0.")
+
+ elif vtype == T_INT:
+ try:
+ if raw_value.strip().lower() in ("none", "null", ""): return None
+ return int(raw_value)
+ except ValueError: raise ValueError(f"Invalid integer value: {raw_value!r}")
+
+ elif vtype == T_FLOAT:
+ try:
+ if raw_value.strip().lower() in ("none", "null", ""): return None
+ return float(raw_value)
+ except ValueError: raise ValueError(f"Invalid float value: {raw_value!r}")
+
+ elif vtype == T_LIST:
+ import json
+ try:
+ val = json.loads(raw_value)
+ if isinstance(val, list): return val
+ else: return [val]
+ except json.JSONDecodeError:
+ parts = [p.strip() for p in raw_value.split(",")]
+ return parts
+
+ elif vtype == T_STR:
+ if raw_value.strip().lower() in ("none", "null"): return None
+ return raw_value
+
+ elif vtype == T_NONE: return None
+
+ return raw_value
+
+def format_value(key, value):
+ if value is None: return "None"
+ if isinstance(value, bool): return "true" if value else "false"
+ if isinstance(value, (int, float)): return str(value)
+ if isinstance(value, list): return str(value)
+ if isinstance(value, bytes): return value.hex()
+ return str(value)
diff --git a/sbapp/sideband/cli/output.py b/sbapp/sideband/cli/output.py
new file mode 100644
index 00000000..015e7155
--- /dev/null
+++ b/sbapp/sideband/cli/output.py
@@ -0,0 +1,211 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
+import re
+import RNS
+import time
+import unicodedata
+
+
+def section(title):
+ bar = "-"*len(title)
+ return f"{title}\n{bar}"
+
+def kv(label, value, indent=0):
+ pad = " " * indent
+ return f"{pad}{label:<30} {value}"
+
+def kv_table(items, indent=0, label_width=30):
+ lines = []
+ pad = " " * indent
+ for label, value in items: lines.append(f"{pad}{str(label):<{label_width}} {str(value)}")
+ return "\n".join(lines) + "\n"
+
+def table(rows, headers=None, indent=0):
+ if not rows and not headers: return ""
+
+ all_rows = []
+ if headers: all_rows.append(headers)
+ all_rows.extend(rows)
+
+ num_cols = max(len(r) for r in all_rows)
+ col_widths = [0] * num_cols
+ for row in all_rows:
+ for i, cell in enumerate(row):
+ col_widths[i] = max(col_widths[i], len(str(cell)))
+
+ pad = " " * indent
+ lines = []
+
+ if headers:
+ header_line = pad + " ".join(str(h).ljust(col_widths[i]) for i, h in enumerate(headers))
+ lines.append(header_line)
+ sep_line = pad + " ".join("-" * col_widths[i] for i in range(num_cols))
+ lines.append(sep_line)
+
+ for row in rows:
+ cells = []
+ for i in range(num_cols):
+ val = str(row[i]) if i < len(row) else ""
+ cells.append(val.ljust(col_widths[i]))
+ lines.append(pad + " ".join(cells))
+
+ return "\n".join(lines) + "\n"
+
+def hexrep(data, delimit=True):
+ return RNS.prettyhexrep(data) if delimit else RNS.hexrep(data, delimit=False)
+
+def timestamp(ts, fmt="%Y-%m-%d %H:%M:%S"):
+ if ts is None: return "-"
+ return time.strftime(fmt, time.localtime(ts))
+
+def time_ago(ts):
+ if ts is None: return "-"
+ delta = time.time() - ts
+ if delta < 0: return "in the future"
+ if delta < 60: return f"{int(delta)}s ago"
+ if delta < 3600: return f"{int(delta/60)}m ago"
+ if delta < 86400: return f"{int(delta/3600)}h ago"
+ return f"{int(delta/86400)}d ago"
+
+def bool_str(val):
+ return "yes" if val else "no"
+
+def none_str(val, default="-"):
+ return default if val is None else str(val)
+
+def truncate(text, length=80):
+ if text is None: return ""
+ s = str(text)
+ if len(s) <= length: return s
+ return s[:length-3] + "..."
+
+def dest_name(sideband, dest_hash):
+ if dest_hash is None: return "-"
+ try:
+ name = sideband.peer_display_name(dest_hash)
+ if name and name != RNS.prettyhexrep(dest_hash):
+ return name
+ except Exception: pass
+ return RNS.prettyhexrep(dest_hash)
+
+def wrap_text(text, width=78, indent=0):
+ if not text: return ""
+ pad = " " * indent
+ lines = []
+ for paragraph in str(text).split("\n"):
+ words = paragraph.split(" ")
+ current = ""
+ for word in words:
+ if len(current) + len(word) + 1 > width - indent:
+ lines.append(pad + current)
+ current = word
+ else: current = current + " " + word if current else word
+ if current: lines.append(pad + current)
+ else: lines.append("")
+ return "\n".join(lines)
+
+def divider(width=78): return "-" * width + "\n"
+
+# Unicode blocks to strip (symbols, dingbats, emoji ranges)
+STRIP_BLOCKS_RE = re.compile(
+ '['
+ '\U0001F600-\U0001F64F' # Emoticons
+ '\U0001F300-\U0001F5FF' # Misc Symbols & Pictographs
+ '\U0001F680-\U0001F6FF' # Transport & Map Symbols
+ '\U0001F700-\U0001F77F' # Alchemical Symbols
+ '\U0001F780-\U0001F7FF' # Geometric Shapes Extended
+ '\U0001F800-\U0001F8FF' # Supplemental Arrows-C
+ '\U0001F900-\U0001F9FF' # Supplemental Symbols & Pictographs
+ '\U0001FA00-\U0001FA6F' # Chess Symbols
+ '\U0001FA70-\U0001FAFF' # Symbols & Pictographs Extended-A
+ '\U0001F1E0-\U0001F1FF' # Flags (iOS/regional indicators)
+ '\u2600-\u26FF' # Misc Symbols (☀, ☁, ☂, etc.)
+ '\u2700-\u27BF' # Dingbats (✂, ✈, ✉, ✌, etc.)
+ '\uFE00-\uFE0F' # Variation Selectors
+ '\U000E0100-\U000E01EF' # Variation Selectors Supplement
+ '\U0001F3FB-\U0001F3FF' # Emoji modifiers (skin tones)
+ ']+',
+ flags=re.UNICODE
+)
+
+# Control characters and zero-width characters to strip
+STRIP_CONTROL_RE = re.compile(
+ '['
+ '\x00-\x08' # C0 controls (NUL-BS)
+ '\x0B\x0C' # VT, FF
+ '\x0E-\x1F' # C0 controls (SO-US)
+ '\x7F-\x9F' # DEL and C1 controls
+ '\u200B-\u200F' # Zero-width chars, LRM, RLM, etc.
+ '\u202A-\u202E' # Bidi embedding controls
+ '\u2060-\u206F' # Format chars (word joiner, etc.)
+ '\uFEFF' # BOM / Zero Width NBSP
+ '\uFFF0-\uFFF8' # Specials
+ ']+',
+ flags=re.UNICODE
+)
+
+# Surrogates and private use areas
+# Shouldn't appear in valid UTF-8, but strip just in case
+STRIP_PRIVATE_RE = re.compile(
+ '['
+ '\uD800-\uDFFF' # Surrogates
+ '\uE000-\uF8FF' # Private Use Area
+ '\uF900-\uFAFF' # CJK Compatibility Ideographs (keep? strip for safety)
+ '\uFE10-\uFE1F' # Vertical Forms
+ '\uFE20-\uFE2F' # Combining Half Marks
+ '\U000F0000-\U000FFFFF' # Supplementary Private Use Area-A
+ '\U00100000-\U0010FFFF' # Supplementary Private Use Area-B
+ ']+',
+ flags=re.UNICODE
+)
+
+def sanitize_name(name):
+ if name is None: return None
+
+ # Convert to string and normalize to NFKC
+ # NFKC: Compatibility decomposition followed by canonical composition
+ # This handles: ① to 1, Ⅰ to I, etc., while keeping composed forms
+ name = str(name)
+ name = unicodedata.normalize('NFKC', name)
+
+ # Build result using category-based filtering
+ result = []
+ for char in name:
+ cat = unicodedata.category(char)
+ cat_prefix = cat[0] if cat else 'C'
+
+ # Allow letters (L*), numbers (N*), and punctuation (P*)
+ if cat_prefix in ('L', 'N', 'P'): result.append(char)
+
+ # Allow space separator, normalize to regular space
+ elif cat == 'Zs': result.append(' ')
+
+ # Convert line/paragraph separators to space
+ elif cat in ('Zl', 'Zp'): result.append(' ')
+
+ # Allow spacing combining marks (Mc) for Indic, Hebrew, etc.
+ elif cat == 'Mc': result.append(char)
+
+ # Allow modifier letters (Lm) - e.g., ʰ, ʱ, ː
+ elif cat == 'Lm': result.append(char)
+
+ # Strip everything else:
+ # - Mn (Nonspacing Mark): diacritics, combining marks (Zalgo)
+ # - Me (Enclosing Mark): enclosing combining marks
+ # - C* (Controls, Format, Surrogates, Private Use, Unassigned)
+ # - S* (Symbols: currency, math, modifiers, other)
+
+ name = ''.join(result)
+
+ # Additional block-based stripping for symbols that categories missed
+ name = STRIP_BLOCKS_RE.sub('', name)
+ name = STRIP_CONTROL_RE.sub('', name)
+ name = STRIP_PRIVATE_RE.sub('', name)
+
+ # Collapse multiple whitespace characters
+ name = re.sub(r'\s+', ' ', name)
+
+ # Strip leading/trailing whitespace
+ name = name.strip()
+
+ return name
\ No newline at end of file
diff --git a/sbapp/sideband/console.py b/sbapp/sideband/console.py
index 9283772b..5f65f3cf 100644
--- a/sbapp/sideband/console.py
+++ b/sbapp/sideband/console.py
@@ -1,98 +1,209 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
-import RNS
+import sys
+import time
+import queue
import threading
-from prompt_toolkit.application import Application
-from prompt_toolkit.document import Document
-from prompt_toolkit.key_binding import KeyBindings
-from prompt_toolkit.layout.containers import HSplit, Window
-from prompt_toolkit.layout.layout import Layout
-from prompt_toolkit.styles import Style
-from prompt_toolkit.widgets import SearchToolbar, TextArea
+
+import RNS
+
+from pathlib import Path
+from .cli import registry
+from .cli.commands import parse_input, resolve_command, format_help, format_command_help
+from .cli.completer import build_completion_tree
+from .cli import cmd_system, cmd_conv, cmd_config, cmd_telemetry, cmd_voice, cmd_net, cmd_plugins
+
+from .console_editor import InlineEditor
sideband = None
-application = None
-output_document = Document(text="", cursor_position=0)
-output_field = None
+_commands_registered = False
+_context = None
+_running = False
+_editor = None
+
+class ConsoleContext:
+ def __init__(self):
+ self.sideband = None
+ self._queue = queue.Queue()
+ self._log = queue.Queue()
+ self._editor_active = False
+ self._stream_mode = False
+ self._lock = threading.Lock()
+
+ def output(self, text=""):
+ text = str(text)
+ with self._lock:
+ if self._editor_active: self._queue.put(text)
+ else: print(text)
+
+ def log(self, text=""):
+ text = str(text)
+ with self._lock:
+ if not self._stream_mode: self._log.put(text)
+ else: print(text)
+
+ def flush(self):
+ with self._lock:
+ while True:
+ try: print(self._queue.get_nowait())
+ except queue.Empty: break
+
+ def flush_log(self):
+ with self._lock:
+ while True:
+ try: print(self._log.get_nowait())
+ except queue.Empty: break
+
+ def clear(self):
+ with self._lock:
+ while True:
+ try: self._queue.get_nowait()
+ except queue.Empty: break
+ if sys.platform != "win32" and sys.stdout.isatty():
+ sys.stdout.write("\x1b[2J\x1b[H")
+ sys.stdout.flush()
+ else: print("\n" + "=" * 40 + "\n")
+
+ def quit(self):
+ global _running
+ _running = False
+ if self.sideband:
+ self.output("Shutting down Sideband...")
+ if self.sideband.owner_app: self.sideband.owner_app.schedule_quit()
+ else: self.sideband.should_persist_data()
+
+ def enter_stream_mode(self): self._stream_mode = True
+
+def _ensure_registered():
+ global _commands_registered
+ if not _commands_registered:
+ cmd_system.register(sideband)
+ cmd_conv.register(sideband)
+ cmd_config.register(sideband)
+ cmd_telemetry.register(sideband)
+ cmd_voice.register(sideband)
+ cmd_net.register(sideband)
+ cmd_plugins.register(sideband)
+ _commands_registered = True
+
+def _receive_output(msg):
+ if _context is not None: _context.output(str(msg))
+
+def set_log(level=None):
+ if level is not None: RNS.loglevel = level
+ if RNS.loglevel == 0: _receive_output("Logging squelched. Use 'log' command to print output to console.")
+
+def _build_console_completer(sideband):
+ tree = build_completion_tree(sideband)
+
+ def completer(text: str) -> list[str]:
+ if not text:
+ tokens = []
+ prefix = ""
+
+ else:
+ tokens = text.split()
+ if text.endswith(" "): prefix = ""
+ else:
+ prefix = tokens[-1] if tokens else ""
+ tokens = tokens[:-1]
+
+ node = tree
+ for token in tokens:
+ if isinstance(node, dict) and token in node: node = node[token]
+ else: return []
+
+ if isinstance(node, dict): return [k for k in node.keys() if k.startswith(prefix)]
+ if hasattr(node, "complete"): return node.complete(prefix)
+
+ return []
+
+ return completer
+
+def _handle_input(text):
+ text = text.strip()
+ if not text: return
+
+ if text == "raw" or text.startswith("raw "):
+ expr = text[4:] if len(text) > 4 else ""
+ cmd = registry.lookup(("raw",))
+ if cmd is not None:
+ try:
+ result = cmd.execute([expr], sideband, _context)
+ if result is not None: _context.output(str(result))
+ except Exception as e: _context.output(f"Error: {e}")
+ else: _context.output("Command 'raw' not available.")
+ return
+
+ tokens = parse_input(text)
+ if not tokens: return
+
+ cmd, args = resolve_command(tokens)
+
+ if cmd is None:
+ _context.output(f"Unknown command: {' '.join(tokens[:3])}")
+ _context.output("Type 'help' for available commands.")
+ return
+
+ error = cmd.validate_args(args)
+ if error is not None:
+ _context.output(error)
+ return
-def attach(target_core, full_screen=False):
- global sideband
- sideband = target_core
- RNS.logdest = RNS.LOG_CALLBACK
- RNS.logcall = receive_output
- console(full_screen=full_screen)
-
-def parse(uin):
- args = uin.split(" ")
- cmd = args[0]
- if cmd == "q" or cmd == "quit": quit_action()
- elif cmd == "clear": cmd_clear(args)
- elif cmd == "raw": cmd_raw(args, uin.replace("raw ", ""))
- elif cmd == "log": cmd_log(args)
- else: receive_output(f"Unknown command: {cmd}")
-
-def cmd_clear(args):
- output_document = output_document = Document(text="", cursor_position=0)
- output_field.buffer.document = output_document
-
-def cmd_raw(args, expr):
- if expr != "" and expr != "raw":
- try: receive_output(eval(expr))
- except Exception as e: receive_output(str(e))
-
-def cmd_log(args):
try:
- if len(args) == 1: receive_output(f"Current loglevel is {RNS.loglevel}")
- else: RNS.loglevel = int(args[1]); receive_output(f"Loglevel set to {RNS.loglevel}")
+ result = cmd.execute(args, sideband, _context)
+ if result is not None: _context.output(str(result))
except Exception as e:
- receive_output("Invalid loglevel: {e}")
+ _context.output(f"Error executing command '{cmd.name()}': {e}")
+ RNS.trace_exception(e) # TODO: Remove
-def set_log(level=None):
- if level: RNS.loglevel = level
- if RNS.loglevel == 0: receive_output("Logging squelched. Use log command to print output to console.")
-
-def quit_action():
- receive_output("Shutting down Sideband...")
- sideband.should_persist_data()
- application.exit()
-
-def receive_output(msg):
- global output_document, output_field
- content = f"{output_field.text}\n{msg}"
- output_document = output_document = Document(text=content, cursor_position=len(content))
- output_field.buffer.document = output_document
-
-def console(full_screen=False):
- global output_document, output_field, application
- search_field = SearchToolbar()
-
- output_field = TextArea(style="class:output-field", text="Sideband console ready")
- input_field = TextArea(
- height=1,
- prompt="> ",
- style="class:input-field",
- multiline=False,
- wrap_lines=False,
- search_field=search_field)
-
- container = HSplit([
- output_field,
- Window(height=1, char="-", style="class:line"),
- input_field,
- search_field])
-
- def accept(buff): parse(input_field.text)
- input_field.accept_handler = accept
-
- kb = KeyBindings()
- @kb.add("c-c")
- @kb.add("c-q")
- def _(event): quit_action()
-
- style = Style([
- ("line", "#004444"),
- ])
-
- application = Application(layout=Layout(container, focused_element=input_field), key_bindings=kb,
- style=style, mouse_support=True, full_screen=full_screen)
+def detach():
+ global _running, _editor
+ _running = False
+ _editor._disable_raw_mode()
+
+def attach(target_core, history_path=None):
+ global sideband, _context, _commands_registered, _running, _editor
+
+ sideband = target_core
+ _commands_registered = False
+ _ensure_registered()
+
+ RNS.logdest = RNS.LOG_CALLBACK
+ RNS.logcall = _receive_output
+
+ _context = ConsoleContext()
+ _context.sideband = sideband
+
+ if history_path:
+ try: history_path = Path(history_path)
+ except Exception: pass
+
+ completer = _build_console_completer(sideband)
+ editor = InlineEditor(history_file=history_path, completer=completer, multiline=False)
+ _editor = editor
+
+ print("Sideband console ready")
set_log()
- application.run()
\ No newline at end of file
+
+ _running = True
+ while _running:
+ if _context._stream_mode:
+ print("[Log streaming active press, hit enter to return to console]")
+ _context.flush_log()
+ try: sys.stdin.readline()
+ except KeyboardInterrupt: pass
+ _context._stream_mode = False
+ continue
+
+ _context._editor_active = True
+ try: line = editor.read("> ")
+ except KeyboardInterrupt: break
+ finally: _context._editor_active = False
+
+ print("\r", end=""); _context.flush()
+ if line.strip(): _handle_input(line)
+
+ RNS.logdest = RNS.LOG_STDOUT
+ RNS.logcall = None
diff --git a/sbapp/sideband/console_editor.py b/sbapp/sideband/console_editor.py
new file mode 100644
index 00000000..c1c6a470
--- /dev/null
+++ b/sbapp/sideband/console_editor.py
@@ -0,0 +1,667 @@
+# Copyright (c) 2026 Mark Qvist
+
+from __future__ import annotations
+
+import os
+import sys
+import termios
+import tty
+import shutil
+from enum import Enum, auto
+from pathlib import Path
+from typing import Optional, Callable
+
+class KeyType(Enum):
+ CHAR = auto()
+ ENTER = auto()
+ BACKSPACE = auto()
+ DELETE = auto()
+ UP = auto()
+ DOWN = auto()
+ LEFT = auto()
+ RIGHT = auto()
+ HOME = auto()
+ END = auto()
+ HOME_FULL = auto() # Ctrl+A
+ END_FULL = auto() # Ctrl+E
+ TAB = auto()
+ SUBMIT = auto() # Ctrl+D / Alt+Enter
+ INTERRUPT = auto() # Ctrl+C
+ CLEAR_BUFFER = auto() # Ctrl+K
+ KILL_LINE_START = auto() # Ctrl+U
+ KILL_WORD_PREV = auto() # Ctrl+W
+ WORD_LEFT = auto() # Ctrl+Left
+ WORD_RIGHT = auto() # Ctrl+Right
+ YANK = auto() # Ctrl+Y
+ TRANSPOSE = auto() # Ctrl+T
+ REFRESH = auto() # Ctrl+L
+ UNKNOWN = auto()
+
+
+class Key:
+ __slots__ = ["type", "char"]
+ def __init__(self, key_type: KeyType, char: str = ""):
+ self.type = key_type
+ self.char = char
+
+ @classmethod
+ def from_char(cls, c: str) -> Key: return cls(KeyType.CHAR, c)
+
+
+class InlineEditor:
+ HISTORY_LIMIT = 1000
+
+ def __init__(self, history_file: Optional[Path] = None, completer: Optional[Callable[[str], list[str]]] = None, multiline: bool = True):
+ self.history_file = history_file
+ self.completer = completer
+ self.multiline = multiline
+
+ self.history: list[str] = []
+ self.history_index: int = 0
+ self._saved_buffer: Optional[list[str]] = None
+ self._saved_row: int = 0
+ self._saved_col: int = 0
+
+ self.buffer: list[str] = [""]
+ self.cursor_row: int = 0
+ self.cursor_col: int = 0
+ self.prompt: str = ""
+ self._prompt_width: int = 0
+
+ self._visual_col: int = 0
+ self._wrapped_view: list[tuple[int, int, str]] = []
+
+ self._orig_termios: Optional[list] = None
+ self._term_width: int = 80
+ self._term_height: int = 24
+
+ self._kill_ring: list[str] = []
+ self._kill_index: int = -1
+
+ self._show_completions: list[str] = []
+
+ if self.history_file: self._load_history()
+
+ def read(self, prompt: str = "") -> str:
+ if sys.platform == "win32" or not sys.stdin.isatty(): return input(prompt)
+ return self._unix_read(prompt)
+
+ def _enable_raw_mode(self):
+ fd = sys.stdin.fileno()
+ self._orig_termios = termios.tcgetattr(fd)
+ tty.setraw(fd, termios.TCSANOW)
+
+ def _disable_raw_mode(self):
+ if self._orig_termios:
+ fd = sys.stdin.fileno()
+ termios.tcsetattr(fd, termios.TCSADRAIN, self._orig_termios)
+ self._orig_termios = None
+
+ def _update_terminal_size(self):
+ try:
+ size = shutil.get_terminal_size()
+ self._term_width = size.columns
+ self._term_height = size.lines
+ except Exception: pass
+
+ def _unix_read(self, prompt: str) -> str:
+ self.prompt = prompt
+ self._prompt_width = len(prompt)
+ self.buffer = [""]
+ self.cursor_row = 0
+ self.cursor_col = 0
+ self._visual_col = self._prompt_width
+ self.history_index = len(self.history)
+ self._saved_buffer = None
+ self._show_completions = []
+
+ self._enable_raw_mode()
+ try:
+ self._render()
+ while True:
+ key = self._read_key()
+ if key.type == KeyType.SUBMIT: return self._finish_read()
+ if not self.multiline and key.type == KeyType.ENTER: return self._finish_read()
+ if key.type == KeyType.INTERRUPT:
+ sys.stdout.write("\n")
+ sys.stdout.flush()
+ raise KeyboardInterrupt
+
+ if key.type == KeyType.UNKNOWN: continue
+
+ self._dispatch(key)
+ self._render()
+
+ finally: self._disable_raw_mode()
+
+ def _finish_read(self) -> str:
+ result = "\n".join(self.buffer)
+ self._add_to_history(result)
+ self._wrapped_view = self._compute_wrapped_view()
+ last_row = len(self.buffer) - 1
+ last_col = len(self.buffer[last_row])
+ end_vis_row, end_vis_col = self._logical_to_visual(last_row, last_col)
+ current_vis_row = getattr(self, "_last_render_row", 0)
+ row_delta = end_vis_row - current_vis_row
+ if row_delta > 0: sys.stdout.write(f"\x1b[{row_delta}B")
+ elif row_delta < 0: sys.stdout.write(f"\x1b[{-row_delta}A")
+ sys.stdout.write(f"\r\x1b[{end_vis_col}C")
+ sys.stdout.write("\n")
+ sys.stdout.flush()
+ return result
+
+ def _read_key(self) -> Key:
+ c = sys.stdin.read(1)
+ if not c: return Key(KeyType.UNKNOWN)
+ o = ord(c)
+ if o == 4: return Key(KeyType.SUBMIT)
+ if o == 3: return Key(KeyType.INTERRUPT)
+ if c in ("\r", "\n"): return Key(KeyType.ENTER)
+ if c == "\t": return Key(KeyType.TAB)
+ if c in ("\x08", "\x7f"): return Key(KeyType.BACKSPACE)
+ if c == "\x1b": return self._read_escape_sequence()
+ if o == 10 and self._orig_termios: return Key(KeyType.SUBMIT)
+ if o == 1: return Key(KeyType.HOME_FULL)
+ if o == 5: return Key(KeyType.END_FULL)
+ if o == 11: return Key(KeyType.CLEAR_BUFFER)
+ if o == 21: return Key(KeyType.KILL_LINE_START)
+ if o == 23: return Key(KeyType.KILL_WORD_PREV)
+ if o == 25: return Key(KeyType.YANK)
+ if o == 20: return Key(KeyType.TRANSPOSE)
+ if o == 12: return Key(KeyType.REFRESH)
+ return Key.from_char(c)
+
+ def _read_escape_sequence(self) -> Key:
+ c1 = sys.stdin.read(1)
+ if c1 in ("\r", "\n"): return Key(KeyType.SUBMIT)
+ if c1 == "[":
+ c2 = sys.stdin.read(1)
+ if c2 == "A": return Key(KeyType.UP)
+ if c2 == "B": return Key(KeyType.DOWN)
+ if c2 == "C": return Key(KeyType.RIGHT)
+ if c2 == "D": return Key(KeyType.LEFT)
+ if c2 == "H": return Key(KeyType.HOME)
+ if c2 == "F": return Key(KeyType.END)
+ if c2 == "1":
+ c3 = sys.stdin.read(1)
+ if c3 == ";":
+ c4 = sys.stdin.read(1)
+ if c4 == "5":
+ c5 = sys.stdin.read(1)
+ if c5 == "C": return Key(KeyType.WORD_RIGHT)
+ if c5 == "D": return Key(KeyType.WORD_LEFT)
+ return Key(KeyType.UNKNOWN)
+ if c2 == "3":
+ if sys.stdin.read(1) == "~": return Key(KeyType.DELETE)
+ return Key(KeyType.UNKNOWN)
+ if c2.isdigit():
+ c3 = sys.stdin.read(1)
+ if c3 == "~":
+ if c2 == "1": return Key(KeyType.HOME)
+ if c2 == "4": return Key(KeyType.END)
+ if c2 == "3": return Key(KeyType.DELETE)
+ return Key(KeyType.UNKNOWN)
+ if c1 == "O":
+ c2 = sys.stdin.read(1)
+ if c2 == "H": return Key(KeyType.HOME)
+ if c2 == "F": return Key(KeyType.END)
+ return Key(KeyType.UNKNOWN)
+
+ def _dispatch(self, key: Key):
+ if key.type != KeyType.TAB: self._show_completions = []
+
+ handlers: dict[KeyType, Callable[[], None]] = {
+ KeyType.CHAR: lambda: self._insert_char(key.char),
+ KeyType.ENTER: self._split_line if self.multiline else lambda: None,
+ KeyType.BACKSPACE: self._backspace,
+ KeyType.DELETE: self._delete,
+ KeyType.LEFT: self._cursor_left,
+ KeyType.RIGHT: self._cursor_right,
+ KeyType.UP: self._cursor_up,
+ KeyType.DOWN: self._cursor_down,
+ KeyType.HOME: self._cursor_home,
+ KeyType.END: self._cursor_end,
+ KeyType.HOME_FULL: self._cursor_home_total,
+ KeyType.END_FULL: self._cursor_end_total,
+ KeyType.CLEAR_BUFFER: self._clear_input,
+ KeyType.KILL_LINE_START: self._kill_line_start,
+ KeyType.KILL_WORD_PREV: self._kill_word_prev,
+ KeyType.WORD_LEFT: self._cursor_word_left,
+ KeyType.WORD_RIGHT: self._cursor_word_right,
+ KeyType.YANK: self._yank,
+ KeyType.TRANSPOSE: self._transpose_chars,
+ KeyType.REFRESH: self._refresh_screen,
+ KeyType.TAB: self._complete,
+ }
+
+ handler = handlers.get(key.type)
+ if handler: handler()
+
+ def _word_start(self) -> int:
+ text = self.buffer[self.cursor_row][:self.cursor_col]
+ pos = text.rfind(" ")
+ return pos + 1 if pos >= 0 else 0
+
+ def _insert_completion(self, completion: str, word_start: int):
+ line = self.buffer[self.cursor_row]
+ self.buffer[self.cursor_row] = line[:word_start] + completion + line[self.cursor_col:]
+ self.cursor_col = word_start + len(completion)
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _format_completions(self, matches: list[str]) -> list[str]:
+ matches = sorted(set(matches))
+ if not matches: return []
+
+ max_len = max(len(m) for m in matches)
+ col_width = max_len + 2
+ n_cols = max(1, (self._term_width - 2) // col_width)
+
+ lines: list[str] = []
+ row: list[str] = []
+ for i, m in enumerate(matches):
+ row.append(m.ljust(col_width))
+ if (i + 1) % n_cols == 0:
+ lines.append(" " + "".join(row).rstrip())
+ row = []
+ if row: lines.append(" " + "".join(row).rstrip())
+
+ return lines
+
+ def _complete(self):
+ if self.completer is None:
+ self._show_completions = []
+ return
+
+ text = self.buffer[self.cursor_row]
+ prefix = text[:self.cursor_col]
+ word_start = self._word_start()
+ word = prefix[word_start:]
+
+ completions = self.completer(prefix)
+ if not completions:
+ self._show_completions = []
+ return
+
+ matches = sorted(set(c for c in completions if c.startswith(word)))
+ if not matches:
+ self._show_completions = []
+ return
+
+ if len(matches) == 1:
+ self._insert_completion(matches[0], word_start)
+ self._show_completions = []
+ return
+
+ common = os.path.commonprefix(matches)
+ if len(common) > len(word):
+ self._insert_completion(common, word_start)
+ self._show_completions = self._format_completions(matches)
+ return
+
+ if self._show_completions: self._show_completions = []
+ else: self._show_completions = self._format_completions(matches)
+
+ def _wrap_line_at_words(self, line: str, avail_width: int) -> list[tuple[int, str]]:
+ if not line: return [(0, "")]
+
+ segments = []
+ start = 0
+ line_len = len(line)
+
+ while start < line_len:
+ end = start + avail_width
+ if end >= line_len:
+ segments.append((start, line[start:]))
+ break
+
+ search_pos = end - 1
+ break_pos = -1
+ while search_pos >= start:
+ if line[search_pos] == " ":
+ break_pos = search_pos
+ break
+ search_pos -= 1
+
+ if break_pos == -1:
+ segments.append((start, line[start:end]))
+ start = end
+ else:
+ segments.append((start, line[start:break_pos]))
+ start = break_pos + 1
+ while start < line_len and line[start] == " ":
+ start += 1
+
+ return segments
+
+ def _compute_wrapped_view(self) -> list[tuple[int, int, str]]:
+ view = []
+ for log_row, line in enumerate(self.buffer):
+ if log_row == 0:
+ avail_width = self._term_width - self._prompt_width
+ if avail_width < 1: avail_width = 1
+ else: avail_width = self._term_width - self._prompt_width
+ wrapped = self._wrap_line_at_words(line, avail_width)
+ for start_col, text in wrapped: view.append((log_row, start_col, text))
+ return view
+
+ def _logical_to_visual(self, log_row: int, log_col: int) -> tuple[int, int]:
+ view = self._wrapped_view
+ for vrow, (v_log_row, start_col, text) in enumerate(view):
+ if v_log_row == log_row:
+ if log_col >= start_col and log_col <= start_col + len(text):
+ vis_col = log_col - start_col + self._prompt_width
+ return (vrow, vis_col)
+ if view:
+ last = view[-1]
+ return (len(view) - 1, self._prompt_width + len(last[2]))
+ return (0, self._prompt_width)
+
+ def _visual_to_logical(self, vis_row: int, vis_col: int) -> tuple[int, int]:
+ if not self._wrapped_view: return (0, 0)
+ if vis_row < 0: vis_row = 0
+ if vis_row >= len(self._wrapped_view): vis_row = len(self._wrapped_view) - 1
+ log_row, start_col, text = self._wrapped_view[vis_row]
+ vis_col -= self._prompt_width
+ if vis_col < 0: vis_col = 0
+ log_col = start_col + vis_col
+ if log_row < len(self.buffer): log_col = min(log_col, len(self.buffer[log_row]))
+ return (log_row, log_col)
+
+ def _render(self):
+ self._update_terminal_size()
+ self._wrapped_view = self._compute_wrapped_view()
+
+ last_row = getattr(self, "_last_render_row", 0)
+ if last_row > 0: sys.stdout.write(f"\x1b[{last_row}A")
+ sys.stdout.write("\r")
+
+ seen = set()
+ for i, (log_row, _start_col, text) in enumerate(self._wrapped_view):
+ if i > 0: sys.stdout.write("\r\n")
+ sys.stdout.write("\x1b[K")
+ first = log_row not in seen
+ seen.add(log_row)
+ if log_row == 0 and first: sys.stdout.write(self.prompt)
+ else: sys.stdout.write(" " * self._prompt_width)
+ sys.stdout.write(text)
+
+ sys.stdout.write("\x1b[J")
+
+ extra_lines = 0
+ if self._show_completions:
+ sys.stdout.write("\r\n")
+ extra_lines += 1
+ for i, line in enumerate(self._show_completions):
+ if i > 0:
+ sys.stdout.write("\r\n")
+ extra_lines += 1
+ sys.stdout.write("\x1b[K")
+ sys.stdout.write(line)
+
+ vis_row, vis_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+ lines_up = len(self._wrapped_view) - vis_row - 1 + extra_lines
+ if lines_up > 0: sys.stdout.write(f"\x1b[{lines_up}A")
+ sys.stdout.write(f"\r\x1b[{vis_col}C")
+ sys.stdout.flush()
+ self._last_render_row = vis_row
+
+ def _insert_char(self, char: str):
+ line = self.buffer[self.cursor_row]
+ self.buffer[self.cursor_row] = line[: self.cursor_col] + char + line[self.cursor_col :]
+ self.cursor_col += 1
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _split_line(self):
+ line = self.buffer[self.cursor_row]
+ new_line = line[self.cursor_col :]
+ self.buffer[self.cursor_row] = line[: self.cursor_col]
+ self.cursor_row += 1
+ self.cursor_col = 0
+ self.buffer.insert(self.cursor_row, new_line)
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _backspace(self):
+ if self.cursor_col > 0:
+ line = self.buffer[self.cursor_row]
+ self.buffer[self.cursor_row] = line[: self.cursor_col - 1] + line[self.cursor_col :]
+ self.cursor_col -= 1
+ elif self.cursor_row > 0:
+ prev = self.buffer[self.cursor_row - 1]
+ curr = self.buffer[self.cursor_row]
+ self.cursor_col = len(prev)
+ self.buffer[self.cursor_row - 1] = prev + curr
+ del self.buffer[self.cursor_row]
+ self.cursor_row -= 1
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _delete(self):
+ line = self.buffer[self.cursor_row]
+ if self.cursor_col < len(line):
+ self.buffer[self.cursor_row] = line[: self.cursor_col] + line[self.cursor_col + 1 :]
+ elif self.cursor_row < len(self.buffer) - 1:
+ nxt = self.buffer[self.cursor_row + 1]
+ self.buffer[self.cursor_row] = line + nxt
+ del self.buffer[self.cursor_row + 1]
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _cursor_left(self):
+ if self.cursor_col > 0:
+ self.cursor_col -= 1
+ elif self.cursor_row > 0:
+ self.cursor_row -= 1
+ self.cursor_col = len(self.buffer[self.cursor_row])
+ if not self._wrapped_view:
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _cursor_right(self):
+ line = self.buffer[self.cursor_row]
+ if self.cursor_col < len(line):
+ self.cursor_col += 1
+ elif self.cursor_row < len(self.buffer) - 1:
+ self.cursor_row += 1
+ self.cursor_col = 0
+ if not self._wrapped_view:
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _cursor_up(self):
+ if not self._wrapped_view:
+ self._update_terminal_size()
+ self._wrapped_view = self._compute_wrapped_view()
+ vis_row, _ = self._logical_to_visual(self.cursor_row, self.cursor_col)
+ if vis_row > 0:
+ new_vis_row = vis_row - 1
+ if new_vis_row < len(self._wrapped_view):
+ _, _, text = self._wrapped_view[new_vis_row]
+ target = self._visual_col
+ min_col = self._prompt_width
+ max_col = self._prompt_width + len(text)
+ if target < min_col: target = min_col
+ if target > max_col: target = max_col
+ self.cursor_row, self.cursor_col = self._visual_to_logical(new_vis_row, target)
+ else: self.cursor_row, self.cursor_col = self._visual_to_logical(new_vis_row, self._visual_col)
+ elif self.cursor_row == 0 and self.cursor_col == 0: self._history_prev()
+ elif self.cursor_row == 0 and self.cursor_col > 0: self.cursor_col = 0
+
+ def _cursor_down(self):
+ if not self._wrapped_view:
+ self._update_terminal_size()
+ self._wrapped_view = self._compute_wrapped_view()
+ vis_row, _ = self._logical_to_visual(self.cursor_row, self.cursor_col)
+ if vis_row < len(self._wrapped_view) - 1:
+ new_vis_row = vis_row + 1
+ if new_vis_row < len(self._wrapped_view):
+ _, _, text = self._wrapped_view[new_vis_row]
+ target = self._visual_col
+ min_col = self._prompt_width
+ max_col = self._prompt_width + len(text)
+ if target < min_col: target = min_col
+ if target > max_col: target = max_col
+ self.cursor_row, self.cursor_col = self._visual_to_logical(new_vis_row, target)
+ else: self.cursor_row, self.cursor_col = self._visual_to_logical(new_vis_row, self._visual_col)
+ elif self.cursor_row == len(self.buffer) - 1 and self.cursor_col >= len(self.buffer[self.cursor_row]): self._history_next()
+ elif self.cursor_row == len(self.buffer) - 1: self._cursor_end()
+
+ def _cursor_home(self):
+ self.cursor_col = 0
+ if not self._wrapped_view: self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _cursor_end(self):
+ self.cursor_col = len(self.buffer[self.cursor_row])
+ if not self._wrapped_view: self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _cursor_home_total(self):
+ self.cursor_row = 0
+ self.cursor_col = 0
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(0, 0)
+
+ def _cursor_end_total(self):
+ if not self.buffer: return
+ self.cursor_row = len(self.buffer) - 1
+ self.cursor_col = len(self.buffer[-1])
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _cursor_word_left(self):
+ if self.cursor_col > 0:
+ line = self.buffer[self.cursor_row]
+ i = self.cursor_col - 1
+ while i >= 0 and not line[i].isalnum(): i -= 1
+ while i >= 0 and line[i].isalnum(): i -= 1
+ self.cursor_col = i + 1
+ elif self.cursor_row > 0:
+ self.cursor_row -= 1
+ self.cursor_col = len(self.buffer[self.cursor_row])
+ if not self._wrapped_view:
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _cursor_word_right(self):
+ line = self.buffer[self.cursor_row]
+ if self.cursor_col < len(line):
+ i = self.cursor_col
+ while i < len(line) and line[i].isalnum(): i += 1
+ while i < len(line) and not line[i].isalnum(): i += 1
+ self.cursor_col = i
+ elif self.cursor_row < len(self.buffer) - 1:
+ self.cursor_row += 1
+ self.cursor_col = 0
+ if not self._wrapped_view:
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _clear_input(self):
+ self.buffer = [""]
+ self.cursor_row = 0
+ self.cursor_col = 0
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(0, 0)
+
+ def _kill_line_start(self):
+ killed = self.buffer[self.cursor_row][: self.cursor_col]
+ self.buffer[self.cursor_row] = self.buffer[self.cursor_row][self.cursor_col :]
+ self.cursor_col = 0
+ if killed:
+ self._kill_ring.append(killed)
+ self._kill_index = len(self._kill_ring) - 1
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _kill_word_prev(self):
+ line = self.buffer[self.cursor_row]
+ if self.cursor_col == 0: return
+ i = self.cursor_col - 1
+ while i >= 0 and not line[i].isalnum(): i -= 1
+ while i >= 0 and line[i].isalnum(): i -= 1
+ killed = line[i + 1 : self.cursor_col]
+ self.buffer[self.cursor_row] = line[: i + 1] + line[self.cursor_col :]
+ self.cursor_col = i + 1
+ if killed:
+ self._kill_ring.append(killed)
+ self._kill_index = len(self._kill_ring) - 1
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _yank(self):
+ if 0 <= self._kill_index < len(self._kill_ring):
+ for c in self._kill_ring[self._kill_index]: self._insert_char(c)
+
+ def _transpose_chars(self):
+ line = self.buffer[self.cursor_row]
+ if self.cursor_col > 0 and self.cursor_col < len(line):
+ self.buffer[self.cursor_row] = (line[: self.cursor_col - 1] + line[self.cursor_col] + line[self.cursor_col - 1] + line[self.cursor_col + 1 :])
+ self.cursor_col += 1
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _refresh_screen(self):
+ sys.stdout.write("\x1b[H\x1b[2J")
+ self._render()
+
+ def _save_current_buffer(self):
+ self._saved_buffer = self.buffer.copy()
+ self._saved_row = self.cursor_row
+ self._saved_col = self.cursor_col
+
+ def _restore_saved_buffer(self):
+ if self._saved_buffer is not None:
+ self.buffer = self._saved_buffer
+ self.cursor_row = self._saved_row
+ self.cursor_col = self._saved_col
+ self._saved_buffer = None
+ self._update_terminal_size()
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _load_history_entry(self, entry: str):
+ if "\n" in entry: self.buffer = entry.split("\n")
+ else: self.buffer = [entry]
+ self.cursor_row = len(self.buffer) - 1
+ self.cursor_col = len(self.buffer[-1])
+ self._update_terminal_size()
+ self._wrapped_view = self._compute_wrapped_view()
+ _, self._visual_col = self._logical_to_visual(self.cursor_row, self.cursor_col)
+
+ def _history_prev(self):
+ if self.history_index == len(self.history): self._save_current_buffer()
+ if self.history_index > 0:
+ self.history_index -= 1
+ self._load_history_entry(self.history[self.history_index])
+
+ def _history_next(self):
+ if self.history_index < len(self.history) - 1:
+ self.history_index += 1
+ self._load_history_entry(self.history[self.history_index])
+ elif self.history_index == len(self.history) - 1:
+ self.history_index = len(self.history)
+ self._restore_saved_buffer()
+
+ def _add_to_history(self, entry: str):
+ if not entry.strip(): return
+ if self.history and self.history[-1] == entry: return
+ self.history.append(entry)
+ if len(self.history) > self.HISTORY_LIMIT: self.history = self.history[-self.HISTORY_LIMIT :]
+ self._save_history()
+
+ def _load_history(self):
+ if not self.history_file or not self.history_file.exists(): return
+ try:
+ text = self.history_file.read_text(encoding="utf-8")
+ self.history = [e for e in text.split("\x00") if e]
+ except Exception: self.history = []
+
+ def _save_history(self):
+ if not self.history_file: return
+ try:
+ self.history_file.parent.mkdir(parents=True, exist_ok=True)
+ self.history_file.write_text("\x00".join(self.history), encoding="utf-8")
+ except Exception: pass
diff --git a/sbapp/sideband/core.py b/sbapp/sideband/core.py
index 7afa40e5..f76c0644 100644
--- a/sbapp/sideband/core.py
+++ b/sbapp/sideband/core.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import RNS
import LXMF
import threading
@@ -108,7 +110,7 @@ class SidebandCore():
CONV_BROADCAST = 0x03
CONV_VOICE = 0x04
- MAX_ANNOUNCES = 24
+ MAX_ANNOUNCES = 48
SERVICE_JOB_INTERVAL = 1
PERIODIC_JOBS_INTERVAL = 60
@@ -203,10 +205,6 @@ class SidebandCore():
self.default_config_template = rns_config
if config_path == None:
- # h_dir = plyer.storagepath.get_home_dir()
- # if type(h_dir) == bytes: h_dir = h_dir.decode("utf-8")
- # self.app_dir = os.path.join(h_dir, ".config", "sideband")
- # if self.app_dir.startswith("file://"): self.app_dir = self.app_dir.replace("file://", "")
self.app_dir = os.path.join(plyer.storagepath.get_home_dir(), ".config", "sideband")
if self.app_dir.startswith("file://"): self.app_dir = self.app_dir.replace("file://", "")
else:
@@ -872,6 +870,21 @@ class SidebandCore():
if self.is_client: self.setstate("wants.settings_reload", True)
RNS.log("Sideband configuration saved", RNS.LOG_DEBUG)
+ def reload_plugins(self):
+ for plugin_name in self.active_telemetry_plugins:
+ try: self.active_telemetry_plugins[plugin_name].stop()
+ except Exception as e: RNS.log(f"Could not stop plugin \"{plugin_name}\": {e}", RNS.LOG_ERROR)
+ for plugin_name in self.active_command_plugins:
+ try: self.active_command_plugins[plugin_name].stop()
+ except Exception as e: RNS.log(f"Could not stop plugin \"{plugin_name}\": {e}", RNS.LOG_ERROR)
+ for plugin_name in self.active_service_plugins:
+ try: self.active_service_plugins[plugin_name].stop()
+ except Exception as e: RNS.log(f"Could not stop plugin \"{plugin_name}\": {e}", RNS.LOG_ERROR)
+ self.active_telemetry_plugins = {}
+ self.active_command_plugins = {}
+ self.active_service_plugins = {}
+ self.__load_plugins()
+
def __load_plugins(self):
plugins_path = self.config["command_plugins_path"]
command_plugins_enabled = self.config["command_plugins_enabled"] == True
@@ -956,8 +969,7 @@ class SidebandCore():
if self.is_standalone and not self.is_daemon: self.owner_app.save_window_config()
def set_active_propagation_node(self, dest):
- if dest == None:
- RNS.log("No active propagation node configured")
+ if dest == None: RNS.log("No active propagation node configured")
else:
try:
if self.message_router:
diff --git a/sbapp/sideband/geo.py b/sbapp/sideband/geo.py
index e950a8c1..64c7ea31 100644
--- a/sbapp/sideband/geo.py
+++ b/sbapp/sideband/geo.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
import time
import mmap
diff --git a/sbapp/sideband/mqtt.py b/sbapp/sideband/mqtt.py
index 57efe001..37100a50 100644
--- a/sbapp/sideband/mqtt.py
+++ b/sbapp/sideband/mqtt.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import RNS
import time
import threading
diff --git a/sbapp/sideband/plugins.py b/sbapp/sideband/plugins.py
index 2de8ac42..be64e11f 100644
--- a/sbapp/sideband/plugins.py
+++ b/sbapp/sideband/plugins.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
class SidebandPlugin():
pass
diff --git a/sbapp/sideband/res.py b/sbapp/sideband/res.py
index 5c528550..68e89e9d 100644
--- a/sbapp/sideband/res.py
+++ b/sbapp/sideband/res.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
sideband_fb_data = [
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x7f, 0xfe, 0x00, 0x00, 0x00,
0x00, 0x00, 0x07, 0xff, 0xff, 0xe0, 0x00, 0x00, 0x00, 0x00, 0x1f, 0xff, 0xff, 0xf8, 0x00, 0x00,
diff --git a/sbapp/sideband/sense.py b/sbapp/sideband/sense.py
index 4d49d909..8a24ea05 100644
--- a/sbapp/sideband/sense.py
+++ b/sbapp/sideband/sense.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
import RNS
import time
diff --git a/sbapp/sideband/voice.py b/sbapp/sideband/voice.py
index ca039472..e9b6efc6 100644
--- a/sbapp/sideband/voice.py
+++ b/sbapp/sideband/voice.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import RNS
import os
import sys
diff --git a/sbapp/ui/__init__.py b/sbapp/ui/__init__.py
index f49e360a..49612e76 100644
--- a/sbapp/ui/__init__.py
+++ b/sbapp/ui/__init__.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
import glob
diff --git a/sbapp/ui/announces.py b/sbapp/ui/announces.py
index 52ed461b..837f3f09 100644
--- a/sbapp/ui/announces.py
+++ b/sbapp/ui/announces.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
diff --git a/sbapp/ui/conversations.py b/sbapp/ui/conversations.py
index 54694a21..74a9ca2c 100644
--- a/sbapp/ui/conversations.py
+++ b/sbapp/ui/conversations.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import RNS
import time
diff --git a/sbapp/ui/guide.py b/sbapp/ui/guide.py
index ff16126e..8103d966 100644
--- a/sbapp/ui/guide.py
+++ b/sbapp/ui/guide.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
diff --git a/sbapp/ui/hardware.py b/sbapp/ui/hardware.py
index cf0b5c8c..947583ae 100644
--- a/sbapp/ui/hardware.py
+++ b/sbapp/ui/hardware.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
diff --git a/sbapp/ui/helpers.py b/sbapp/ui/helpers.py
index 3f2f3fca..9909ef0c 100644
--- a/sbapp/ui/helpers.py
+++ b/sbapp/ui/helpers.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
from kivy.utils import get_color_from_hex
from kivymd.color_definitions import colors
from kivy.uix.screenmanager import ScreenManager, Screen
diff --git a/sbapp/ui/keys.py b/sbapp/ui/keys.py
index cd62b9cc..50dcd79f 100644
--- a/sbapp/ui/keys.py
+++ b/sbapp/ui/keys.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
diff --git a/sbapp/ui/layouts.py b/sbapp/ui/layouts.py
index b56db15a..2b0ffe76 100644
--- a/sbapp/ui/layouts.py
+++ b/sbapp/ui/layouts.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
root_layout = """
#: import NoTransition kivy.uix.screenmanager.NoTransition
#: import SlideTransition kivy.uix.screenmanager.SlideTransition
diff --git a/sbapp/ui/map.py b/sbapp/ui/map.py
index 5ba25cbf..26be4f27 100644
--- a/sbapp/ui/map.py
+++ b/sbapp/ui/map.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import os
import RNS
import time
diff --git a/sbapp/ui/messages.py b/sbapp/ui/messages.py
index 7c3e18ac..05edafb8 100644
--- a/sbapp/ui/messages.py
+++ b/sbapp/ui/messages.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
import LXMF
diff --git a/sbapp/ui/objectdetails.py b/sbapp/ui/objectdetails.py
index 4ee2056d..441f0c43 100644
--- a/sbapp/ui/objectdetails.py
+++ b/sbapp/ui/objectdetails.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
diff --git a/sbapp/ui/telemetry.py b/sbapp/ui/telemetry.py
index 320ac886..acbaf71f 100644
--- a/sbapp/ui/telemetry.py
+++ b/sbapp/ui/telemetry.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
diff --git a/sbapp/ui/utilities.py b/sbapp/ui/utilities.py
index 4fcb6978..e4003deb 100644
--- a/sbapp/ui/utilities.py
+++ b/sbapp/ui/utilities.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
diff --git a/sbapp/ui/voice.py b/sbapp/ui/voice.py
index bae133bc..099ab62d 100644
--- a/sbapp/ui/voice.py
+++ b/sbapp/ui/voice.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2026 Mark Qvist - See LICENSE.md and README.md
+
import time
import RNS
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────